在 Spark Streaming Python 中将 RDD 转换为 Dataframe答案

【问题标题】：Convert RDD to Dataframe in Spark Streaming Python在 Spark Streaming Python 中将 RDD 转换为 Dataframe
【发布时间】：2019-05-14 16:58:01
【问题描述】：

我正在尝试在 Spark Streaming 中将 RDD 转换为 DataFrame。我正在遵循以下流程。

socket_stream = ssc.socketTextStream("localhost", 9999)
def convert_to_df(rdd):
    schema = StructType([StructField("text", StringType(), True)])
    df =spark.createDataFrame(rdd, schema = schema)
    df.show(10)

socket_stream.foreachRDD(convert_to_df)

我通过套接字nc -lk 9999提供输入

如果我输入“hello world”作为我的输入，它会显示以下错误

StructType can not accept object 'hello world' in type <class 'str'>

预期输出

+-------=-+
|text     |
+---------+
hello world
+---------+

【问题讨论】：

标签： python apache-spark apache-spark-sql spark-streaming

【解决方案1】：

由于您使用RDD[str]，您应该提供匹配类型。对于原子值，它是对应的AtomicType

from pyspark.sql.types import StringType, StructField, StructType

rdd = sc.parallelize(["hello world"])
spark.createDataFrame(rdd, StringType())

或其字符串描述：

spark.createDataFrame(rdd, "string")

如果你想先使用StructTypeconvert data to tuples：

schema = StructType([StructField("text", StringType(), True)])

spark.createDataFrame(rdd.map(lambda x: (x, )), schema)

当然，如果您只想将每个批次转换为 DataFrame，那么一直使用结构化流更有意义：

lines = (spark
    .readStream
    .format("socket")
    .option("host", "localhost")
    .option("port", 9999)
    .load())

【讨论】：

【解决方案2】：

试试ArrayType(StringType())

否则，由于您只有一列，请尝试直接将架构指定为

df =spark.createDataFrame(rdd, StringType())

检查 udf 以获取 pyspark，因为您需要为 spark 声明 udf

【讨论】：