有一個帶有時間戳欄位的資料框，如下所示：
| 時間戳 | ID | 版本 |
|---|---|---|
| 2022-01-01 01:02:00.000 | 1 | 2 |
| 2022-01-01 05:12:00.000 | 1 | 2 |
我創建了一個 Glue 作業，ApplyMapping用于將資料保存到新的 S3 位置。目前，我已經通過在可視化編輯器中選擇這些欄位來添加id和版本磁區，并且我的資料使用以下結構保存：id=1/version=2/我想決議時間戳并提取日期值，因此檔案系統結構將是id=1/version=2/dt=2022-01-01/. 但是，在可視化編輯器中，我只能選擇時間戳，不能對欄位執行任何操作。我猜我需要更改代碼，但我不確定如何。
代碼：
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)
# Script generated for node S3 bucket
S3bucket_node1 = glueContext.create_dynamic_frame.from_options(
format_options={},
connection_type="s3",
format="parquet",
connection_options={"paths": ["s3://my-data"], "recurse": True},
transformation_ctx="S3bucket_node1",
)
# Script generated for node ApplyMapping
ApplyMapping_node2 = ApplyMapping.apply(
frame=S3bucket_node1,
mappings=[
("timestamp", "timestamp", "timestamp", "timestamp"),
("id", "string", "id", "string"),
("version", "string", "version", "string"),
],
transformation_ctx="ApplyMapping_node2",
)
# Script generated for node S3 bucket
S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
frame=ApplyMapping_node2,
connection_type="s3",
format="glueparquet",
connection_options={
"path": "s3://target-data",
"partitionKeys": ["id", "version"],
},
format_options={"compression": "gzip"},
transformation_ctx="S3bucket_node3",
)
job.commit()
uj5u.com熱心網友回復：
使用地圖類。
將此方法添加到您的腳本中
def AddDate(rec):
ts = str(rec["timestamp"])
rec["dt"] = ts[:10]
return rec
ApplyMapping在該步驟之后插入地圖變換。
Mapped_dyF = Map.apply(frame = ApplyMapping_node2, f = AddDate)
更新寫入 S3 步驟，注意對frame和的更改partitionKeys。
S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
frame=Mapped_dyF,
connection_type="s3",
format="glueparquet",
connection_options={
"path": "s3://target-data",
"partitionKeys": ["id", "version", "dt"],
},
format_options={"compression": "gzip"},
transformation_ctx="S3bucket_node3",
)
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/485798.html