时间戳

首页 > 解决方案 > Spark：根据 s3 文件中的字段动态生成查询

scala - Spark：根据 s3 文件中的字段动态生成查询

问题描述

过度简化的场景： 在 s3 文件中生成每月数据的过程。每个月运行的字段数可能不同。基于 s3 中的这些数据，我们将数据加载到表中，然后我们手动（因为每次运行中的字段数量可能会随着添加或删除几列而改变）针对少数指标运行 SQL。对此有更多的计算/转换数据，但要让初学者我展示用例的更简单版本。

方法：考虑到无模式的性质，因为每次运行时 s3 文件中的字段数量可能会有所不同，添加/删除少量字段，这需要每次在 SQL 中手动更改，我计划探索 Spark/Scala，这样我们就可以直接从s3中读取，根据字段动态生成SQL。

查询： 如何在 scala/spark-SQL/dataframe 中实现这一点？s3 文件仅包含每次运行所需的字段。因此，从 s3 读取动态字段没有问题，因为它由数据帧处理。问题是我们如何生成 SQL 数据帧-API/spark-SQL 代码来处理。

我可以通过数据框读取 s3 文件并将数据框注册为 createOrReplaceTempView 以编写 SQL，但我认为在下次运行期间在 s3 中添加新字段期间手动更改 spark-SQL 没有帮助。动态生成 sql 的最佳方法是什么/处理问题的更好方法是什么？

用例 1：

dataframe: customer,1st_month_count（这里dataframe直接指向s3，只有需要的属性）

--sample code
SELECT customer,sum(month_1_count)
FROM dataframe
GROUP BY customer

--Dataframe API/SparkSQL
dataframe.groupBy("customer").sum("month_1_count").show()

dataframe: customer,month_1_count,month_2_count) （这里dataframe直接指向s3，只有需要的属性）

--Sample SQL
SELECT customer,sum(month_1_count),sum(month_2_count)
FROM dataframe
GROUP BY customer

--Dataframe API/SparkSQL
dataframe.groupBy("customer").sum("month_1_count","month_2_count").show()

我是 Spark/Scala 的新手，如果您能提供指导以便我进一步探索，将会很有帮助。

标签： scalaapache-sparkapache-spark-sql

解决方案

听起来您想对出现在数据框架构中的新列一遍又一遍地执行相同的操作？这有效：

from pyspark.sql import functions

#search for column names you want to sum, I put in "month"

column_search = lambda col_names: 'month' in col_names

#get column names of temp dataframe w/ only the columns you want to sum

relevant_columns = original_df.select(*filter(column_search, original_df.columns)).columns

#create dictionary with relevant column names to be passed to the agg function

columns = {col_names: "sum" for col_names in relevant_columns}

#apply agg function with your groupBy, passing in columns dictionary

grouped_df = original_df.groupBy("customer").agg(columns)

#show result

grouped_df.show()

一些重要的概念可以帮助您学习：

  1. DataFrames 具有存储在列表中的数据属性：dataframe.columns

  2. 可以将函数应用于列表以创建新列表，如“column_search”

  3. Agg 函数接受字典中的多个表达式，如此处所述，这是我传递给“列”的内容

  4. Spark 是惰性的，因此在您执行 show() 之类的操作之前，它不会更改数据状态或执行操作。这意味着写出临时数据帧以使用数据帧的一个元素，如我所做的那样并不昂贵，即使如果您习惯于 SQL，它可能看起来效率低下。

推荐阅读