AWS Glue ETL Spark- 字符串到时间戳

Posted 2023-02-16

【中文标题】AWS Glue ETL Spark- 字符串到时间戳

【英文标题】：AWS Glue ETL Spark- string to timestamp

【发布时间】：2021-05-16 02:37:02

【问题描述】：

我正在尝试通过 AWS Glue ETL 作业将我的 CSV 转换为 Parquet。同时，我愿意将我的 datetime 列（字符串）转换为 Athena 可以识别的时间戳格式。 （Athena 认出了这个 yyyy-MM-dd HH:mm:ss）

我浏览并应用了许多建议，但未能成功。

能否请您告诉我应该导入哪个库，并为特定行应用脚本？以下代码是 AWS Glue 建议从 CSV 转换为 Parquet 的代码，并且似乎可以针对我的日期时间转换目的进行自定义。

提前致谢。

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 = "partition_db", table_name = "test_folder", transformation_ctx = "datasource0")

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("col0", "string", "col0", "string"), ("col1", "string", "col1", "string"), ("col2", "string", "col2", "string"), ("col3", "string", "col3", "string"), ("col4", "string", "col4", "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://commercialanalytics/future_partition/test_folder_parquet", format = "parquet", transformation_ctx = "datasink4")
job.commit()

【参考方案1】：

您可以创建一个函数并在 Map 类中调用它。

import pandas as pd

def parse_date(df):
    dt = pd.to_datetime(df["col_name"]).dt.strftime('%Y-%m-%d %H:%M:%S.%f') # Replace col_name with the column name
    return dt

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("col0", "string", "col0", "string"), ("col1", "string", "col1", "string"), ("col2", "string", "col2", "string"), ("col3", "string", "col3", "string"), ("col4", "string", "col4", "string")], transformation_ctx = "applymapping1")
custommapping1 = Map.apply(frame = applymapping1 , f = parse_date, transformation_ctx = "custommapping1")

另一种选择是转换为 Spark Dataframes 并使用 spark.sql(....) 查询

【讨论】：

欢迎来到 SO。如果答案有帮助，请随时投票或接受——这将有助于其他开发人员:)

【参考方案2】：

使用火花数据框。我发现这是最简单的

df= datasource0.toDF()

from pyspark.sql.functions import from_unixtime, unix_timestamp, col
df= df.withColumn(col(columnname),from_unixtime(unix_timestamp(col(columnname),"dd/MM/yyyy hh.mm")))