pyspark - 如何在 RDD 上运行过滤器功能并一次性使用不同的数据帧模式创建输出

Question

我正在阅读带有火花的专有二进制格式（rosbags），需要进行一些反序列化。完成后，我将获得具有固定数量的不同模式的数据。我想编写输出文件，每个不同的模式都有一个。

通过按类型过滤和反序列化，我设法为每个模式创建了一个数据帧，但是所有原始数据都被重复读取。

下面的示例演示了json.loads()用于输入和collect()输出的问题。

import json
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.types import *

def transform(data):
    print("transforming", data)
    return json.loads(data[1])

def filter_by_type(data, type_):
    print("filtering %s == %s" % (data[0], type_))
    return data[0] == type_

spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("ERROR")

dd = sc.parallelize([
    ['type1', '"string1"'],
    ['type2', '2'],
])

print(spark.createDataFrame(dd.filter(lambda x: filter_by_type(x, "type1")).map(transform), StringType()).collect())
print(spark.createDataFrame(dd.filter(lambda x: filter_by_type(x, "type2")).map(transform), IntegerType()).collect())

输出

filtering type1 == type1
transforming ['type1', '"string1"']
filtering type2 == type1
[Row(value='string1')]
filtering type1 == type2
filtering type2 == type2
transforming ['type2', '2']
[Row(value=2)]

标签： pysparkpyspark-sql