如何使用胶水将存储在 s3 中的 json 文件转换为 csv？

Question

我有一些 json 文件存储在 s3 中，我需要在它们所在的文件夹中将它们转换为 csv 格式。

目前我正在使用胶水将它们映射到 athena，但是，正如我所说，现在我需要将它们映射到 csv。

是否可以使用胶水作业来做到这一点？

我试图了解胶水作业是否可以爬入我的 s3 文件夹目录，将它找到的所有 json 文件转换为 csv（作为新文件）。

如果不可能，是否有任何 aws 服务可以帮助我做到这一点？

编辑1：

这是我正在尝试运行的当前代码

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext

sc = SparkContext()
glueContext = GlueContext(sc)

inputGDF = glueContext.create_dynamic_frame_from_options(connection_type = "s3", connection_options = {"paths": ["s3://agco-sa-dfs-dv/dealer-data"]}, format = "json")
outputGDF = glueContext.write_dynamic_frame.from_options(frame = inputGDF, connection_type = "s3", connection_options = {"path": "s3://agco-sa-dfs-dv/dealer-data"}, format = "csv")

作业运行没有错误，但 s3 文件夹上似乎没有发生任何事情。 我假设代码将从 /dealer-data 获取 json 文件并转换为与 csv 相同的文件夹。我可能错了。

EDIT2：

好的，我几乎让它按照我需要的方式工作了。

问题是，创建动态框架仅适用于带有文件的文件夹，而不适用于带有带有文件的子文件夹的文件夹。

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

sc = SparkContext()
glueContext = GlueContext(sc)

inputGDF = glueContext.create_dynamic_frame_from_options(connection_type = "s3", connection_options = {"paths": ["s3://agco-sa-dfs-dv/dealer-data/installations/3555/2019/2"]}, format = "json")

outputGDF = glueContext.write_dynamic_frame.from_options(frame = inputGDF, connection_type = "s3", connection_options = {"path": "s3://agco-sa-dfs-dv/dealer-data/installations/3555/2019/2/bla.csv"}, format = "csv")

上述方法有效，但仅适用于该目录 (../2) 有没有办法读取给定文件夹和子文件夹的所有文件？

Answer 1

对于S3 connection，您应该将recurse 选项设置为True：

inputGDF = glueContext.create_dynamic_frame_from_options(
    connection_type = "s3", 
    connection_options = {
        "paths": ["s3://agco-sa-dfs-dv/dealer-data/installations/3555/2019/2"],
        "recurse" : True
    }, 
    format = "json
)