Pyspark - 规范化数据框答案

【问题标题】：Pyspark - normalize a dataframePyspark - 规范化数据框
【发布时间】：2021-11-13 16:44:49
【问题描述】：

我有这个 API 响应：

{
  '02/09/2021': {
    'ABC': {
      'emp': 'A1',
      'value': '12421'
    },
    'DEF': {
      'emp': 'D1',
      'value': '3345'
    },
    'GHI': {
      'emp': 'G2',
      'value': '260048836600'
    },
    'JKL': {
      'emp': 'J1',
      'value': '66654654'
    }
  }
}

并希望以这种格式规范化表格：

CODE | EMP | VALUE        | DATE
========================================
ABC  | A1  | 12421        | 02/09/2021
DEF  | D1  | 3445         | 02/09/2021
GHI  | G2  | 260048836600 | 02/09/2021
JKL  | J1  | 66654654     | 02/09/2021

我试着做一个爆炸，但我做不到，我怎样才能得到这个结果？

复制它：

import json

api_response = {'02/09/2021':{'ABC':{'emp':'A1','value':'12421'},'DEF':{'emp':'D1','value':'3345'},'GHI':{'emp':'G2','value':'260048836600'},'JKL':{'emp':'J1','value':'66654654'}}}

rdd = spark.sparkContext.parallelize([json.dumps(api_response)])
input_df = spark.read.json(rdd)

【问题讨论】：

你读得怎么样？您的 JSON 似乎无效
我用的是pyspark，这是一个例子……
是的，但我不知道如何阅读它。通过这两个步骤更新您的问题，首先从运行[dataframe].limit(3).collect() 的数据框中收集一些数据并获取架构[dataframe].schema.simpleString()，其中 [dataframe] 是您的变量名。复制起来会更容易
@Kafels 完成，谢谢！

标签： python dataframe pyspark apache-spark-sql

【解决方案1】：

将您的 api 响应解析为数据框的更简单方法是首先转换结果，删除关键日期并将其放入正文中。要将其转换为表格，需要使用stack：

import json

def transform_api_content(api_response):
  for key, value in api_response.items():
    value['DATE'] = key
    yield json.dumps(value)


api_response = {'02/09/2021':{'ABC':{'emp':'A1','value':'12421'},'DEF':{'emp':'D1','value':'3345'},'GHI':{'emp':'G2','value':'260048836600'},'JKL':{'emp':'J1','value':'66654654'}}}
input_df = spark.read.json(sc.parallelize(transform_api_content(api_response)))

# Hard-coded
stack = 'stack(4, "ABC", ABC.emp, ABC.value, "DEF", DEF.emp, DEF.value, "GHI", GHI.emp, GHI.value, "JKL", JKL.emp, JKL.value) AS (CODE, EMP, VALUE)'
output_df = input_df.selectExpr(stack, 'DATE')
output_df.show(truncate=False)

# Dynamic
def stack_columns(dataframe):
  struct_cols = [column for column, schema in dataframe.dtypes if schema.startswith('struct')]
  stack_expr = ['"{0}", {0}.emp, {0}.value'.format(column) for column in struct_cols]
  return 'stack({length}, {stack}) AS (CODE, EMP, VALUE)'.format(length=len(struct_cols), stack=','.join(stack_expr))


output_df = input_df.selectExpr(stack_columns(input_df), 'DATE')
output_df.show(truncate=False)

输出

+----+---+------------+----------+
|CODE|EMP|VALUE       |DATE      |
+----+---+------------+----------+
|ABC |A1 |12421       |02/09/2021|
|DEF |D1 |3345        |02/09/2021|
|GHI |G2 |260048836600|02/09/2021|
|JKL |J1 |66654654    |02/09/2021|
+----+---+------------+----------+

【讨论】：