python之根据某个键值 (pyspark) 从 RDD 创建多个 Spark DataFrame_编程设计

我有一些包含 JSON 对象的文本文件(每行一个对象)。示例:

{"a": 1, "b": 2, "table": "foo"} 
{"c": 3, "d": 4, "table": "bar"} 
{"a": 5, "b": 6, "table": "foo"} 
...

我想根据表名将文本文件的内容解析成Spark DataFrames。所以在上面的例子中，我会有一个数据帧用于“foo”，另一个数据帧用于“bar”。我已经使用以下 (pyspark) 代码将 JSON 行分组到 RDD 内的列表中:

text_rdd = sc.textFile(os.path.join("/path/to/data", "*")) 
tables_rdd = text_rdd.groupBy(lambda x: json.loads(x)['table'])

这会产生一个包含具有以下结构的元组列表的 RDD:

RDD[("foo", ['{"a": 1, "b": 2, "table": "foo"}', ...], 
    ("bar", ['{"c": 3, "d": 4, "table": "bar"}', ...]]

如何将这个 RDD 分解为每个表键的 DataFrame？

编辑:我试图在上面阐明，在包含表格信息的单个文件中有多行。我知道我可以在我创建的“groupBy”RDD 上调用 .collectAsMap，但我知道这会占用我的驱动程序中相当大的 RAM。我的问题是:有没有办法在不使用 .collectAsMap 的情况下将“groupBy”RDD 分解为多个数据帧？

请您参考如下方法:

您可以有效地将其拆分为 Parquet 分区: 首先，我们将其转换为数据框:

text_rdd = sc.textFile(os.path.join("/path/to/data", "*")) 
df = spark.read.json(text_rdd) 
df.printSchema() 
    root 
     |-- a: long (nullable = true) 
     |-- b: long (nullable = true) 
     |-- c: long (nullable = true) 
     |-- d: long (nullable = true) 
     |-- table: string (nullable = true)

现在我们可以写了:

df.write.partitionBy('table').parquet([output directory name])

如果您列出 [输出目录名称] 的内容，您将看到与 table 的不同值一样多的分区:

hadoop fs -ls [output directory name] 
 
    _SUCCESS 
    table=bar/ 
    table=foo/

如果你只想保留每个表的列，你可以这样做(假设只要表出现在文件中，列的完整列表就会出现)

import ast 
from pyspark.sql import Row 
table_cols = spark.createDataFrame(text_rdd.map(lambda l: ast.literal_eval(l)).map(lambda l: Row( 
        table = l["table"],  
        keys = sorted(l.keys()) 
    ))).distinct().toPandas() 
table_cols = table_cols.set_index("table") 
table_cols.to_dict()["keys"] 
 
    {u'bar': [u'c', u'd', u'table'], u'foo': [u'a', u'b', u'table']}