如何在pyspark中动态从JSON文件中传递Dataframe列值？

Question

我正在使用下面的代码创建数据帧，它正在按预期工作。在

我的数据集是'testdata'：

1|123
2|223
3|323
4|423

from pyspark.sql import SQLContext,SparkSession
from pyspark.sql import Row
spark = SparkSession.builder.appName("test").getOrCreate()
sc = spark.sparkContext
sqlContext = SQLContext(sc)
df_transac = spark.createDataFrame(sc.textFile("testdata").map( lambda x: x.split("|")[:2]).map( lambda r: Row( testA = r[0],testb = r[1])))
df_transac.show()

+---------+---------+
|   testA | testB   |
+---------+---------+
|      123|        1|
|      223|        2|
|      323|        3|
|      423|        4|
+---------+---------+

在创建时间testA的dataframe上面，testB是硬编码的列名，但我想从json获取这些值，所以我尝试了以下方法。 我的json文件testjson.json:

^{pr2}$

然后我尝试通过执行下面的代码来创建数据帧， 但它的投掷错误。在

import json
from pyspark.sql import SQLContext,SparkSession
from pyspark.sql import Row
with open(testjson.json) as spec_data:
    jsn = json.load(spec_data)
spark = SparkSession.builder.appName("test").getOrCreate()
sc = spark.sparkContext
sqlContext = SQLContext(sc)
df_transac = spark.createDataFrame(sc.textFile("testdata").map( lambda x: x.split("|")[:2]).map( lambda r: Row( jsn['column1'] = r[0], jsn['column2'] = r[1])))

引发错误，例如：SyntaxError:keyword不能是表达式。在

我的预期产出是：

+-----------+-----------+
|testcolumn1|testcolumn2|
+-----------+-----------+
|          1|        123|
|          2|        223|
|          3|        323|
|          4|        423|
+-----------+-----------+

请帮忙，我怎样才能做到这一点。在

提前谢谢。在

Answer 1

df_transac = spark.createDataFrame(sc.textFile("testdata").map( lambda x: x.split("|")[:2]).map( lambda r: Row( jsn['column1'] = r[0], jsn['column2'] = r[1])))

使用下面的代码来解决问题，而不是上面的代码。在

^{pr2}$

Answer 2

如例外所述-不能将表达式用作关键字，因此：

Row( jsn['column1'] = r[0], jsn['column2'] = r[1])

不是有效的Python代码。在

您可以使用替代构造函数，然后应用参数：

^{pr2}$

但总的来说，最好是

tmp = spark.read.option("delimiter", "|").csv("testdata")
df = tmp.select(tmp.columns[2:]).toDF(jsn['column1'], jsn['column2'])