将数据附加到空数据框答案

【问题标题】：Appending data to an empty dataframe将数据附加到空数据框
【发布时间】：2018-10-13 19:09:39
【问题描述】：

我正在创建一个空数据框，然后尝试将另一个数据框附加到该数据框。事实上，我想根据即将到来的 RDD 的数量动态地将许多数据帧附加到最初的空数据帧。

如果我将值分配给另一个第三个数据帧，则 union() 函数可以正常工作。

val df3=df1.union(df2)

但我想继续附加到我创建的初始数据帧（空），因为我想将所有 RDD 存储在一个数据帧中。然而，下面的代码没有显示正确的计数。好像根本就没有追加

df1.union(df2)

df1.count() // this shows 0 although df2 has some data and that is shown if I assign to third datafram.

如果我执行以下操作（我得到重新分配错误，因为 df1 是 val。如果我将其更改为 var 类型，我得到 kafka multithreading not safe 错误。

df1=d1.union(df2)

知道如何将所有动态创建的数据帧添加到一个最初创建的数据帧吗？

【问题讨论】：

您通过合并应用的转换未保存。为了节省你必须做的任务。这就是您在数据框中保存转换的方式，我建议您查看 foldLeft 选项以获得所需的联合
Ramesh 关于分配是正确的。我也可能建议不要将具有不同模式的数据框联合起来。 Spark 在此阶段不会合并模式，您最终会得到错误列中的数据。
谢谢，但目前我正在使用相同的架构。你能给我在这种情况下做 foldLeft 的例子吗？赞赏。
嗨 Omer，对数据集的任何转换都会创建一个新的数据框，因此如果您不通过链接进行附加，那么它对您没有帮助。考虑了解数据框及其不变性特性以避免任何混淆。
就像@MichelLemay 提到的，您需要注意架构。

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

【解决方案1】：

不确定这是否是您要找的！

# Import pyspark functions
from pyspark.sql.types import StructType, StructField, IntegerType, StringType 

# Define your schema
field = [StructField("Col1",StringType(), True), StructField("Col2", IntegerType(), True)]
schema = StructType(field)

# Your empty data frame
df = spark.createDataFrame(sc.emptyRDD(), schema)

l = []

for i in range(5):
    # Build and append to the list dynamically
    l = l + [([str(i), i])]

    # Create a temporary data frame similar to your original schema
    temp_df = spark.createDataFrame(l, schema)

    # Do the union with the original data frame
    df = df.union(temp_df)
df.show()

【讨论】：

【解决方案2】：

DataFrames 和其他分布式数据结构是不可变的，因此对它们进行操作的方法总是返回新对象。没有附加，没有修改，也没有 ALTER TABLE 等效项。

如果我将其更改为 var 类型，则会出现 kafka multithreading not safe 错误。

没有实际代码是不可能给你一个确定的答案，但也不太可能与union代码有关。

不正确的内部实现会导致许多已知的 Spark 错误（SPARK-19185、SPARK-23623 仅列举几个）。

【讨论】：