在本文中,我們將回顧如何從一個變量創建一個Apache火花DataFrame包含一個JSON字符串或一個Python字典。

創建一個火花DataFrame JSON字符串

1. JSON內容變量添加到一個列表。

   % scala scala.collection.mutable進口。ListBuffer val json_content1 = " {“json_col1”:“你好”,“json_col2”: 32}“val json_content2 = " {json_col1:‘你好’,‘json_col2’:‘世界’}“var json_seq = new ListBuffer [String] () json_seq + = json_content1 json_seq + = json_content2

2. 從列表中創建一個火花數據集。

   % scala val json_ds = json_seq.toDS ()

3. 使用spark.read.json解析引發數據集。

   % scala val df = spark.read.json (json_ds)顯示(df)

結合示例代碼

這些示例代碼塊結合前麵的步驟為單獨的例子。Python和Scala樣本執行相同的任務。

% python json_content1 = " {“json_col1”:“你好”,“json_col2”: 32}“json_content2 = " {json_col1:‘你好’,‘json_col2’:‘世界’}“json_list = [] json_list.append (json_content1) json_list.append (json_content2) df = spark.read.json (sc.parallelize (json_list))顯示(df)

% scala scala.collection.mutable進口。ListBuffer val json_content1 = " {“json_col1”:“你好”,“json_col2”: 32}“val json_content2 = " {json_col1:‘你好’,‘json_col2’:‘世界’}“var 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)顯示(df)

提取字符串列從DataFrame和解析JSON數據

1. 從DataFrame選擇JSON列,將它轉換成一個抽樣的類型抽樣(行)。

   % scala 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”,“令牌”)val row_rdd = test_df.select (col (json))。抽樣/ /選擇JSON列並將它轉換為抽樣。

2. 轉換抽樣(行)來抽樣(字符串)。

   % scala val string_rdd = row_rdd.map (_.mkString (", "))

3. 使用spark.read.json解析抽樣(字符串)。

   % scala val df1 = spark.read.json (string_rdd)顯示(df1)

結合示例代碼

這個示例代碼塊結合前麵的步驟為一個例子。

% scala 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”,“令牌”)val row_rdd = test_df.select (col (json))。抽樣val string_rdd = row_rdd.map (_.mkString (", ")) val df1 = spark.read.json (string_rdd)顯示(df1)

創建一個火花DataFrame從Python字典

1. 檢查數據類型和確認的詞典類型。

   % python jsonDataDict = {" job_id”: 33100年,“run_id”: 1048560,“number_in_job”: 1、“狀態”:{“life_cycle_state”:“等待”,“state_message”:“等待集群”},“任務”:{" notebook_task ": {“notebook_path”:“/用戶/ user@m.eheci.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@m.eheci.com”、“run_name”:“我的測試工作”,“run_page_url”:“https://testurl.m.eheci.com的工作/ 33100 /跑步/ 1”、“run_type”:“SUBMIT_RUN”} (jsonDataDict)類型

2. 使用json.dumpsPython字典轉換為一個JSON字符串。

   % python導入json jsonData = json.dumps (jsonDataDict)

3. JSON內容添加到一個列表。

   % python jsonDataList = [] jsonDataList.append (jsonData)

4. 將列表轉換成使用抽樣和解析它spark.read.json。

   % python jsonRDD = sc.parallelize (jsonDataList) df = spark.read.json (jsonRDD)顯示(df)

結合示例代碼

這些示例代碼塊結合前麵的步驟為一個例子。

% python jsonDataDict = {" job_id”: 33100年,“run_id”: 1048560,“number_in_job”: 1、“狀態”:{“life_cycle_state”:“等待”,“state_message”:“等待集群”},“任務”:{" notebook_task ": {“notebook_path”:“/用戶/ user@m.eheci.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@m.eheci.com”、“run_name”:“我的測試工作”,“run_page_url”:“https://testurl.m.eheci.com的工作/ 33100 /跑步/ 1”、“run_type”:“SUBMIT_RUN”}類型(jsonDataDict)進口json jsonData = json.dumps (jsonDataDict) jsonDataList = [] jsonDataList.append (jsonData) jsonRDD = sc.parallelize (jsonDataList) df = spark.read.json (jsonRDD)顯示(df)

例如筆記本電腦

檢查解析JSON字符串或Python字典示例筆記本。