aws glue / pyspark - как программно создать таблицу Athena с помощью Glue - PullRequest
Новая
       29

aws glue / pyspark - как программно создать таблицу Athena с помощью Glue

0 голосов
/ 31 мая 2019

Я запускаю скрипт в AwsGlue, который загружает данные из s3, выполняет некоторые преобразования и сохраняет результаты в S3.Я пытаюсь добавить еще один шаг к этой рутине.Я хочу создать новую таблицу в существующей базе данных в Афине.

Я не могу найти подобный пример в документации AWS.Результаты просто записаны в S3 в примерах, с которыми я столкнулсяВозможно ли это в Клее?

Вот пример кода.Как его следует изменить, чтобы создать таблицу Athena с выходными результатами? 

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()

1 Ответ

1 голос
/ 31 мая 2019

Я могу придумать два способа сделать это.Один использует sdk, чтобы получить ссылку на API-интерфейс athena и использовать его для выполнения запроса с оператором создания таблицы , как показано в этом сообщении в блоге

Альтернативный способ, который может бытьболее интересным является использование Glue API для создания сканера для вашей корзины S3 и последующего запуска сканера.

При втором подходе ваша таблица заносится в каталог, и вы можете использовать ее не только из Афины, , но также из EMR или спектра Redshift.

...