向pyspark Dataframe中添加新行

Question

我对pyspark非常新手，但对pandas很熟悉。 我有一个pyspark的Dataframe。

# instantiate Spark
spark = SparkSession.builder.getOrCreate()

# make some test data
columns = ['id', 'dogs', 'cats']
vals = [
     (1, 2, 0),
     (2, 0, 1)
]

# create DataFrame
df = spark.createDataFrame(vals, columns)

想要添加新的行（4、5、7），以便输出：

df.show()
+---+----+----+
| id|dogs|cats|
+---+----+----+
|  1|   2|   0|
|  2|   0|   1|
|  4|   5|   7|
+---+----+----+

- Roushan

Answer 1

正如thebluephantom所说，联合是前进的道路。我只是回答你的问题，给你一个pyspark的例子：

# if not already created automatically, instantiate Sparkcontext
spark = SparkSession.builder.getOrCreate()

columns = ['id', 'dogs', 'cats']
vals = [(1, 2, 0), (2, 0, 1)]

df = spark.createDataFrame(vals, columns)

newRow = spark.createDataFrame([(4,5,7)], columns)
appended = df.union(newRow)
appended.show()

请查看databricks常见问题解答： https://kb.databricks.com/data/append-a-row-to-rdd-or-dataframe.html

- cronoik

Answer 2

要将行添加到数据框中，可以使用collect方法。collect()函数将数据框转换为列表，然后可以直接将数据附加到列表中，再将列表转换回数据框。

我的Spark数据框称为df，如下所示：

+---+----+------+
| id|name|gender|
+---+----+------+
|  1|   A|     M|
|  2|   B|     F|
|  3|   C|     M|
+---+----+------+

使用 collect 将此数据帧转换为列表

collect_df = df.collect()
print(collect_df)

[Row(id=1, name='A', gender='M'),
 Row(id=2, name='B', gender='F'),
 Row(id=3, name='C', gender='M')]

向此列表添加新行

collect_df.append({"id" : 5, "name" : "E", "gender" : "F"})
print(collect_df)

[Row(id=1, name='A', gender='M'),
 Row(id=2, name='B', gender='F'),
 Row(id=3, name='C', gender='M'),
 {'id': 5, 'name': 'E', 'gender': 'F'}]

将这个列表转换为数据帧

added_row_df = spark.createDataFrame(collect_df)
added_row_df.show()

+---+----+------+
| id|name|gender|
+---+----+------+
|  1|   A|     M|
|  2|   B|     F|
|  3|   C|     M|
|  5|   E|     F|
+---+----+------+

- bhargav3vedi

Answer 3

以下是一个使用union的代码块示例，您需要根据自己的情况进行调整：

val dummySchema = StructType(
StructField("phrase", StringType, true) :: Nil)
var dfPostsNGrams2 = spark.createDataFrame(sc.emptyRDD[Row], dummySchema)
for (i <- i_grams_Cols) {
    val nameCol = col({i})
    dfPostsNGrams2 = dfPostsNGrams2.union(dfPostsNGrams.select(explode({nameCol}).as("phrase")).toDF )
 }

使用DF的并集就是正确的方法。

- thebluephantom

---

这个例子有点偏离主题，但它是关于 UNION 的。 - thebluephantom

Answer 4

另一种选择是利用分区parquet格式，并为每个要附加的数据帧添加一个额外的parquet文件。这样，您可以创建（数百、数千、数百万）个parquet文件，当您稍后读取目录时，spark将只读取它们的联合。

此示例使用pyarrow

请注意，我还展示了如何编写一个未分区的单个parquet（example.parquet），如果您已经知道要放置单个parquet文件的位置。

import pyarrow.parquet as pq
import pandas as pd

headers=['A', 'B', 'C']

row1 = ['a1', 'b1', 'c1']
row2 = ['a2', 'b2', 'c2']

df1 = pd.DataFrame([row1], columns=headers)
df2 = pd.DataFrame([row2], columns=headers)

df3 = df1.append(df2, ignore_index=True)


table = pa.Table.from_pandas(df3)

pq.write_table(table, 'example.parquet', flavor='spark')
pq.write_to_dataset(table, root_path="test_part_file", partition_cols=['B', 'C'], flavor='spark')

# Adding a new partition (B=b2/C=c3


row3 = ['a3', 'b3', 'c3']
df4 = pd.DataFrame([row3], columns=headers)

table2 = pa.Table.from_pandas(df4)
pq.write_to_dataset(table2, root_path="test_part_file", partition_cols=['B', 'C'], flavor='spark')

# Add another parquet file to the B=b2/C=c2 partition
# Note this does not overwrite existing partitions, it just appends a new .parquet file.
# If files already exist, then you will get a union result of the two (or multiple) files when you read the partition
row5 = ['a5', 'b2', 'c2']
df5 = pd.DataFrame([row5], columns=headers)
table3 = pa.Table.from_pandas(df5)
pq.write_to_dataset(table3, root_path="test_part_file", partition_cols=['B', 'C'], flavor='spark')

之后读取输出

from pyspark.sql import SparkSession

spark = (SparkSession
         .builder
         .appName("testing parquet read")
         .getOrCreate())

df_spark = spark.read.parquet('test_part_file')
df_spark.show(25, False)

你应该看到类似于这样的东西

+---+---+---+
|A  |B  |C  |
+---+---+---+
|a5 |b2 |c2 |
|a2 |b2 |c2 |
|a1 |b1 |c1 |
|a3 |b3 |c3 |
+---+---+---+

如果您再次端对端地运行相同的内容，您应该会看到像这样的重复项（因为之前的所有Parquet文件仍然存在，Spark会将它们合并）。

+---+---+---+
|A  |B  |C  |
+---+---+---+
|a2 |b2 |c2 |
|a5 |b2 |c2 |
|a5 |b2 |c2 |
|a2 |b2 |c2 |
|a1 |b1 |c1 |
|a1 |b1 |c1 |
|a3 |b3 |c3 |
|a3 |b3 |c3 |
+---+---+---+

- stephen meckstroth