Pyspark拆分字符串类型的火花数据框

Question

我正在使用spark(批处理，而不是流)从kafka topic中读取数据来创建spark dataframe。我想使用spark将这个数据帧加载到cassandra。Dataframe是字符串格式，如下所示。

root |-value:string(nullable = true)

+--------------------+
|value               |
+--------------------+
|"1,Visa,6574"       |
|"3,Visa,6574"       |
|"4,MasterCard,6574" |
|"5,MasterCard,6574" |
|"8,Maestro,8372"    |
+--------------------+

我尝试使用'，'分隔符拆分数据帧记录，并形成新的数据帧，我可以将其数据到cassandra。

创建了如下的火花DF。

df = spark \
.read \
.format("kafka") \
.option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS_CONS) \
.option("subscribe", KAFKA_TOPIC_NAME_CONS) \
.option("startingOffsets", "earliest") \
.load()
df2=df.selectExpr("CAST(value AS STRING)")
df2.printSchema()

我尝试使用'，'拆分数据。

split_col=split(df2['value'],',')
df3=df2.withColumn('Name1',split_col.getItem(0))
df3=df2.withColumn('Name2',split_col.getItem(1))
df3=df2.withColumn('Name3',split_col.getItem(2))

上面的代码没有给出预期的结果，我觉得

根| -- 值： 字符串 （可以为空 = 真） | -- 名称3： 字符串 （可以为空 = 真）

+-------------------+-----+
|value              |Name3|
+-------------------+-----+
|"1,Visa,6574"      |6574"|
|"3,Visa,6574"      |6574"|
|"4,MasterCard,6574"|6574"|
|"5,MasterCard,6574"|6574"|
|"8,Maestro,8372"   |8372"|
+-------------------+-----+

我想得到如下的put:

+-------------------+----------+------+
|Name1              |Name2     |Name3 |
+-------------------+----------+------+
| 1                 |Visa      |6574  |
| 3                 |Visa      |6574  |
| 4                 |MasterCard|6574  |
| 5                 |MasterCard|6574  |
| 8                 |Maestro   |8372  |
+-------------------+----------+------+

请帮忙！！

Answer 1

您的解决方案完全可以。唯一的问题是在执行拆分并用于下一步之后，df2和df3的赋值。完成第一次拆分后，您分配给df3，但对于后续拆分，您只使用df2。因此，只有第三个split语句被spark评估。

解决方案明智，要么在最后一次拆分之前不要分配给新变量

df3 = df2.withColumn('Name1', f.split('value', ',').getItem(0)).\
                 withColumn('Name2', f.split('value', ',').getItem(1)).\
                 withColumn('Name3', f.split('value', ',').getItem(2))

df3.show()
+-----------------+-----+----------+-----+
|            value|Name1|     Name2|Name3|
+-----------------+-----+----------+-----+
|      1,Visa,6574|    1|      Visa| 6574|
|      3,Visa,6574|    3|      Visa| 6574|
|4,MasterCard,6574|    4|MasterCard| 6574|
|5,MasterCard,6574|    5|MasterCard| 6574|
|   8,Maestro,8372|    8|   Maestro| 8372|
+-----------------+-----+----------+-----+

或者在下一个split中使用赋值变量(除非必要，否则不鼓励使用这种方式)

df3 = df2.withColumn('Name1', f.split('value', ',').getItem(0))

df3 = df3.withColumn('Name2', f.split('value', ',').getItem(1))

df3 = df3.withColumn('Name3', f.split('value', ',').getItem(2))

df3.show()
+-----------------+-----+----------+-----+
|            value|Name1|     Name2|Name3|
+-----------------+-----+----------+-----+
|      1,Visa,6574|    1|      Visa| 6574|
|      3,Visa,6574|    3|      Visa| 6574|
|4,MasterCard,6574|    4|MasterCard| 6574|
|5,MasterCard,6574|    5|MasterCard| 6574|
|   8,Maestro,8372|    8|   Maestro| 8372|
+-----------------+-----+----------+-----+