在本文中，我们将回顾如何从包含JSON字符串或Python字典的变量创建Apache Spark DataFrame。
从JSON字符串创建一个Spark DataFrame
- 将变量中的JSON内容添加到列表中。
%scala import scala.collection.mutable.ListBuffer val json_content1 = "{'json_col1': 'hello'， 'json_col2': 'world'}" val json_content2 = "{'json_col1': 'hello'， 'json_col2': 'world'}" var json_seq = new ListBuffer[String]() json_seq += json_content1 json_seq += json_content2 - 从列表中创建一个Spark数据集。
%scala val json_ds = json_seq.toDS()
- 使用spark.read.json来解析Spark数据集。
%scala val df= spark.read.json(json_ds)
组合示例代码
这些示例代码块将前面的步骤组合成单独的示例。Python和Scala示例执行相同的任务。
%python json_content1 = "{'json_col1': 'hello'， 'json_col2': 'world'}" json_list = [] json_list.append(json_content1) json_list.append(json_content2) df = spark.read.json(sc.parallelize(json_list)) display(df)
%scala import scala.collect .mutable.ListBuffer val json_content1 = "{'json_col1': 'hello'， 'json_col2': 'world'}" val json_seq = new ListBuffer[String]() json_seq += json_content1 json_seq += json_content2 val json_ds = json_seq. tods () val df= spark.read.json(json_ds) display(df)
从DataFrame中提取带有JSON数据的字符串列并解析它
- 从DataFrame中选择JSON列并将其转换为相应类型的RDD抽样(行)．
导入org.apache.spark.sql.functions。_ val test_df = Seq((“1”,“{“json_col1”:“你好”,“json_col2”:32}”,“1.0”),(“1”,“{json_col1:‘你好’,‘json_col2’:‘世界’}”,“1.0”))。toDF("row_number"， "json"， "token") val row_rdd = test_df.select(col("json"))。rdd //只选择JSON列并将其转换为rdd。 - 转换抽样(行)来抽样(字符串)．
%scala val string_rdd = row_rdd.map(_.mkString("，")) - 使用spark.read.json要解析抽样(字符串)．
%scala val df1= spark.read.json(string_rdd) display(df1)
组合示例代码
这个示例代码块将前面的步骤组合成一个示例。
导入org.apache.spark.sql.functions。_ val test_df = Seq((“1”,“{“json_col1”:“你好”,“json_col2”:32}”,“1.0”),(“1”,“{json_col1:‘你好’,‘json_col2’:‘世界’}”,“1.0”))。toDF("row_number"， "json"， "token") val row_rdd = test_df.select(col("json"))。rdd val string_rdd = row_rdd.map(_.mkString("，")) val df1= spark.read.json(string_rdd) display(df1)
从Python字典中创建一个Spark DataFrame
- 检查数据类型并确认它是字典类型。
%python jsonDataDict = {"job_id":33100，"run_id":1048560，"number_in_job":1，"state":{"life_cycle_state":"PENDING"，"state_message":"Waiting for .集群”},“任务”:{" notebook_task ":{“notebook_path”:“/用户/ user@www.neidfyre.com/path/test_notebook”}},“cluster_spec ": {" new_cluster ":{“spark_version”:“4.3.x-scala2.11”,“属性”:{“类型”:“fixed_node”、“记忆”:“8 g”},“enable_elastic_disk”:“false”,“num_workers”:1}},“cluster_instance ":{“cluster_id”:“0000 - 000000 wares10”},“start_time”:1584689872601,“setup_duration”:0,”execution_duration”:0,”cleanup_duration”:0,”creator_user_name”:“user@www.neidfyre.com”、“run_name”:“我的测试工作”、“run_page_url”:“https://testurl.www.neidfyre.com作业/ 33100 /运行/ 1”、“run_type”:“SUBMIT_RUN“}类型(jsonDataDict) - 使用json.dumps将Python字典转换为JSON字符串。
%python import json jsonData = json.dump (jsonDataDict)
- 将JSON内容添加到列表中。
%python jsonDataList = [] jsonDataList.append(jsonData)
- 将列表转换为RDD并使用spark.read.json．
%python jsonRDD = sc.parallelize(jsonDataList) df = spark.read.json(jsonRDD) display(df)
组合示例代码
这些示例代码块将前面的步骤组合成一个示例。
%python jsonDataDict = {"job_id":33100，"run_id":1048560，"number_in_job":1，"state":{"life_cycle_state":"PENDING"，"state_message":"Waiting for .集群”},“任务”:{" notebook_task ":{“notebook_path”:“/用户/ user@www.neidfyre.com/path/test_notebook”}},“cluster_spec ": {" new_cluster ":{“spark_version”:“4.3.x-scala2.11”,“属性”:{“类型”:“fixed_node”、“记忆”:“8 g”},“enable_elastic_disk”:“false”,“num_workers”:1}},“cluster_instance ":{“cluster_id”:“0000 - 000000 wares10”},“start_time”:1584689872601,“setup_duration”:0,”execution_duration”:0,”cleanup_duration”:0,”creator_user_name”:“user@www.neidfyre.com”、“run_name”:“我的测试job"，"run_page_url":"https://testurl.www.neidfyre.com#job/33100/run/1"，"run_type":"SUBMIT_RUN"} type(jsonDataDict) import json jsonData = json.dumps(jsonDataDict) jsonDataList = [] jsondatelist .append(jsonData) jsonRDD = sc.parallelize(jsonDataList) df = spark.read.json(jsonRDD) display(df)