3

我想使用AWS Glue將一些csv數據轉換爲orc。
ETL任務我創建生成以下PySpark腳本:使用AWS內部的SQL Glue 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數據(從位置其中雅典娜表tests.test_glue_csv點)並輸出到s3://glue/output/

如何在此腳本中插入一些SQL操作?

感謝

回答

3

你應該首先從動態幀

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

這裏創建一個臨時視圖/表,dyf是你的動態幀。

然後,用你的火花對象上應用SQL查詢

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