在 AWS Glue pySpark 脚本中使用 SQL答案

【问题标题】：use SQL inside AWS Glue pySpark script在 AWS Glue pySpark 脚本中使用 SQL
【发布时间】：2018-01-30 13:12:50
【问题描述】：

我想使用 AWS Glue 将一些 csv 数据转换为 orc。
我创建的 ETL 作业生成了以下 PySpark 脚本：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "tests", table_name = "test_glue_csv", transformation_ctx = "datasource0")

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("id", "int", "id", "int"), ("val", "string", "val", "string")], transformation_ctx = "applymapping1")

resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")

dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")

datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://glue/output"}, format = "orc", transformation_ctx = "datasink4")
job.commit()

它获取 csv 数据（从 Athena 表 tests.test_glue_csv 指向的位置）并输出到s3://glue/output/。

如何在这个脚本中插入一些 SQL 操作？

谢谢

【问题讨论】：

标签： amazon-web-services pyspark pyspark-sql amazon-athena aws-glue

【解决方案1】：

你可以使用toDF()

df = datasource0.toDF() df.printSchema()

【讨论】：

【解决方案2】：

你应该首先从你的动态框架中创建一个临时视图/表格

dyf.toDF().createOrReplaceTempView("view_dyf")

这里，dyf 是您的动态框架。

然后，使用您的 spark 对象对其应用 sql 查询

sqlDF = spark.sql("select * from view_dyf")
sqlDF.show()

【讨论】：