小编rya*_*yan的帖子

我想逐行处理文件。但是，这些文件具有不同的行分隔符："\r","\n"或"\r\n". 我不知道他们使用的是哪种操作系统，也不知道他们来自哪种操作系统。

我有两个解决方案：

1. 使用 bash 命令将这些分隔符转换为"\n".

   cat file |
   tr '\r\n' '\n' |
   tr '\r' '\n' |
   ruby process.rb
   

   Run Code Online (Sandbox Code Playgroud)

2. 读取整个文件并 gsub 这些分隔符

   text=File.open('xxx.txt').read
   text.gsub!(/\r\n?/, "\n")
   text.each_line do |line|
     do some thing
   end
   

   Run Code Online (Sandbox Code Playgroud)

但是当文件很大时，第二种解决方案并不好。见参考。有没有其他 ruby​​ 惯用且有效的解决方案？

假设,我有一个json文件,其中包含以下结构中的行:

{
 "a": 1,
 "b": {
       "bb1": 1,
       "bb2": 2
      }
}

我想更改密钥的值bb1或添加一个新密钥,如:bb3.目前,我使用spark.read.json将json文件作为DataFrame加载到spark中,并使用df.rdd.map将RDD的每一行映射到dict.然后,更改嵌套键值或添加嵌套键并将dict转换为行.最后,将RDD转换为DataFrame.工作流程如下:

def map_func(row):
  dictionary = row.asDict(True)
  adding new key or changing key value
  return as_row(dictionary) # as_row convert dict to row recursively

df = spark.read.json("json_file")
df.rdd.map(map_func).toDF().write.json("new_json_file")

这对我有用.但我担心转换DataFrame - > RDD(Row - > dict - > Row) - > DataFrame会导致效率下降.有没有其他方法可以满足这种需求,但不能以效率为代价？

我使用的最终解决方案是使用withColumn并动态构建b的模式.首先,我们可以通过以下方式获取b_schemafrom df架构:

b_schema = next(field['type'] for field in df.schema.jsonValue()['fields'] if field['name'] == 'b')

在那之后,b_schema是dict,我们可以通过以下方式添加新字段:

b_schema['fields'].append({"metadata":{},"type":"string","name":"bb3","nullable":True})

然后,我们可以通过以下方式将其转换为StructType:

new_b = StructType.fromJson(b_schema)

在map_func中,我们可以将Row转换为dict并填充新字段:

def …