Создайте разделенные данные, используя AWS Glue и сохраните в s3 - PullRequest
Новая
       7

Создайте разделенные данные, используя AWS Glue и сохраните в s3

0 голосов
/ 14 января 2020 
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
from pyspark.sql.functions import col,year,month,dayofmonth,to_date,from_unixtime

## @params: [JOB_NAME]
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 = "db_name", table_name = "table_name", transformation_ctx = "datasource0")

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

df = applymapping1.toDF()

repartitioned_with_new_columns_df = applymapping1.select("*")
    .withColumn("date_col", to_date(from_unixtime(col("dateRegistered"))))
    .withColumn("year", year(col("date_col")))
    .withColumn("month", month(col("date_col")))
    .withColumn("day", dayofmonth(col("date_col")))
    .drop(col("date_col"))
    #.repartition(1)

dyf = DynamicFrame.fromDF(repartitioned_with_new_columns_df, glueContext, "enriched")

datasink = glueContext.write_dynamic_frame.from_options(
    frame = dyf, 
    connection_type = "s3", 
    connection_options = {
        "path": "bucket-path", 
        "partitionKeys": ["year", "month", "day"]
    }, 
    format = "json", 
    transformation_ctx = "datasink")

job.commit()

У меня есть сценарий выше, и я не могу понять, почему он не работает, или если это даже правильный путь.

Может кто-то, пожалуйста, просмотрите и дайте мне знать, что я делаю неправильно?

Цель здесь состоит в том, чтобы ежедневно выполнять это задание, записать эту таблицу, разбитую на разделы, как указано выше, и сохранить ее в s3 либо json, либо в паркет.

...