使用 Pyspark 从 ORC 中提取数据答案

【问题标题】：Extract data out of ORC using Pyspark使用 Pyspark 从 ORC 中提取数据
【发布时间】：2018-03-14 19:00:01
【问题描述】：

我有一个 ORC 文件，我可以使用 Pyspark 2.2.0 将其读入 DataFrame

from pyspark.context import SparkContext
from pyspark.sql.types import *
from pyspark.sql.functions import *

df = spark.read.orc("s3://leadid-sandbox/krish/lead_test/")

上面的 df 有一个如下的模式

root
 |-- item: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)

样本数据看起来像这样（只是一个样本数据而不是整个数据集）

item
{http_Accept-Language={"s":"en-US"}, Win64={"n":"1"}, 
geoip_region={"s":"FL"}, Platform={"s":"Win7"}, geoip_postal_code=
{"s":"33432"}, JavaApplets={"n":"1"}, http_Accept={"s":"*/*"}, 
Version={"s":"11.0"}, Cookies={"n":"1"}, Platform_Version=
{"s":"6.1"}, http_Content-Type={"s":"application/x-www-form-
urlencoded"}}
{http_Accept-Language={"s":"en-US"}, Win64={"n":"1"}, IFrames=
{"n":"1"}, geoip_region={"s":"CA"}, Platform={"s":"Win7"}, Parent=
{"s":"IE 11.0"}, http_Dnt={"n":"1"}}

所以我像下面这样分解了“项目”

expDf = df.select(explode("item"))

上面的 DataFrame 有下面的架构，当我做一个 show(2) 有下面的细节

root
 |-- key: string (nullable = false)
 |-- value: string (nullable = true)

+------------+----------+
| key| value|
+------------+----------+
|geoip_region|
{
    "s": "FL"
}
|
| Tables|
{
   "n": "1"
}
|
+------------+----------+

如何从这个 DataFrame 中选择数据？我尝试了不同的方法，但没有用。所以我需要值为'FL'的'geoip_region'等等。任何帮助表示赞赏。

【问题讨论】：

标签： python apache-spark pyspark orc

【解决方案1】：

我不确定您的完整用例，但如果它只是关于访问“项目”中的键和值，您可以使用以下示例代码来完成：

row = df.select(df.item).collect()

上面的行会给你一个Row对象的列表，比如[Row(item={http_Accept-Language={"s":"en-US"}, Win64={"n":"1"},....})]

然后选择行内的所有值，您可以这样做：row_item = row[0]['item']

row_item['http_Accept'] 将允许您访问u"{"s":"en-US"}"

eval(row_item['http_Accept']) 会给你一个字典，你可以从中获取它的键、值

我刚刚概述了这个过程，可以用循环编写它以获取迭代中的所有键/值。

【讨论】：

您好 Joshi 感谢您对此进行调查并回答，我不太清楚如何将输出键 'http_Accept' 与值 'en-US' 映射。我认为这就是您在 'eval(row_item['http_Accept'])' 上所说的，但我不太了解这部分。
我的错，试试这个： row = df.select(df.item).collect() , lang_dict = eval(row[0]['item']['http_Accept-Language']) , lang_dict['s']

【解决方案2】：

感谢 Joshi 的回复，由于某种原因，我的代码中出现 row[0] not found 错误，我在 AWS Glue 环境中运行它可能是一个原因。

我用下面的代码得到了我想要的。

# Creating a DataFrame of the raw file

df = spark.read.orc("s3://leadid-sandbox/krish/lead_test/")

# Creating a temp view called Leads for the above dataFrame
df.createOrReplaceTempView("leads")

# Extracting the data using normal SQL from the above created Temp 
  View
tblSel = spark.sql("SELECT get_json_object(item['token'], '$.s') as 
token, get_json_object(item['account_code'], '$.s') as account_code 
from leads").show()

【讨论】：