aws glue / pyspark - 如何使用 Glue 以编程方式创建 Athena table
aws glue / pyspark - how to create Athena table programmatically using Glue
我是 运行 AwsGlue 中的一个脚本，它从 s3 加载数据，进行一些转换并将结果保存到 S3。我正在尝试在此例程中再添加一个步骤。我想在 Athena 的现有数据库中创建一个新的 table。
我在 AWS 文档中找不到任何类似的示例。在我遇到的例子中，结果只是写到 S3。
这在 Glue 中可能吗？
有一些代码示例。应该如何修改以创建具有输出结果的 Athena table？
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql import SparkSession
from pyspark.context import SparkContext
from pyspark.sql.functions import *
from pyspark.sql import SQLContext
from pyspark.sql.types import *
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 = "dataset", table_name = "table_1", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("id", "long", "id", "long"), ("description", "string", "description", "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://..."}, format = "parquet", transformation_ctx = "datasink4")
*create Athena table with the output results*
job.commit()
我可以想到两种方法来做到这一点。一种是使用 sdk 获取对 athena API 的引用，并使用它通过 create table 语句 as seen at this blog post
执行查询
另一种可能更有趣的方法是为您的 S3 存储桶使用 Glue API 到 create a crawler，然后执行爬虫。
使用第二种方法，您的 table 已编目，您不仅可以从 athena、but also 从 EMR 或红移光谱中使用它。
我是 运行 AwsGlue 中的一个脚本，它从 s3 加载数据，进行一些转换并将结果保存到 S3。我正在尝试在此例程中再添加一个步骤。我想在 Athena 的现有数据库中创建一个新的 table。
我在 AWS 文档中找不到任何类似的示例。在我遇到的例子中，结果只是写到 S3。 这在 Glue 中可能吗？
有一些代码示例。应该如何修改以创建具有输出结果的 Athena table？
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql import SparkSession
from pyspark.context import SparkContext
from pyspark.sql.functions import *
from pyspark.sql import SQLContext
from pyspark.sql.types import *
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 = "dataset", table_name = "table_1", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("id", "long", "id", "long"), ("description", "string", "description", "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://..."}, format = "parquet", transformation_ctx = "datasink4")
*create Athena table with the output results*
job.commit()
我可以想到两种方法来做到这一点。一种是使用 sdk 获取对 athena API 的引用，并使用它通过 create table 语句 as seen at this blog post
执行查询另一种可能更有趣的方法是为您的 S3 存储桶使用 Glue API 到 create a crawler，然后执行爬虫。
使用第二种方法，您的 table 已编目，您不仅可以从 athena、but also 从 EMR 或红移光谱中使用它。