在 AWS Glue Python Shell 中查询 Athena 表

Question

AWS Glue 中引入了 Python Shell 作业。他们提到：

您现在可以使用 Python shell 作业，例如，将 SQL 查询提交到服务，例如... Amazon Athena ...

好的。我们有一个从 Athena 表 here 读取数据的示例：

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

glueContext = GlueContext(SparkContext.getOrCreate())

persons = glueContext.create_dynamic_frame.from_catalog(
             database="legislators",
             table_name="persons_json")
print("Count: ", persons.count())
persons.printSchema()
# TODO query all persons

但是，它使用 Spark 而不是 Python Shell。没有通常可用于 Spark 作业类型的此类库，我有一个错误：

ModuleNotFoundError：没有名为“awsglue.transforms”的模块

如何重写上面的代码以使其在 Python Shell 作业类型中可执行？

Answer 1

问题是，Python Shell 类型有自己有限的一组内置 libraries。

我只是使用Boto 3 查询数据和Pandas 将其读入数据框来实现我的目标。

这里是sn-p的代码：

import boto3
import pandas as pd

s3 = boto3.resource('s3')
s3_client = boto3.client('s3')
athena_client = boto3.client(service_name='athena', region_name='us-east-1')
bucket_name = 'bucket-with-csv'
print('Working bucket: {}'.format(bucket_name))

def run_query(client, query):
    response = client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={ 'Database': 'sample-db' },
        ResultConfiguration={ 'OutputLocation': 's3://{}/fromglue/'.format(bucket_name) },
    )
    return response

def validate_query(client, query_id):
    resp = ["FAILED", "SUCCEEDED", "CANCELLED"]
    response = client.get_query_execution(QueryExecutionId=query_id)
    # wait until query finishes
    while response["QueryExecution"]["Status"]["State"] not in resp:
        response = client.get_query_execution(QueryExecutionId=query_id)

    return response["QueryExecution"]["Status"]["State"]

def read(query):
    print('start query: {}\n'.format(query))
    qe = run_query(athena_client, query)
    qstate = validate_query(athena_client, qe["QueryExecutionId"])
    print('query state: {}\n'.format(qstate))

    file_name = "fromglue/{}.csv".format(qe["QueryExecutionId"])
    obj = s3_client.get_object(Bucket=bucket_name, Key=file_name)
    return pd.read_csv(obj['Body'])

time_entries_df = read('SELECT * FROM sample-table')

Answer 2

SparkContext 在 Glue Python Shell 中不可用。因此，您需要依赖 Boto3 和 Pandas 来处理数据检索。但是使用 boto3 查询 Athena 并轮询 ExecutionId 以检查查询执行是否完成会带来很多开销。

最近 awslabs 发布了一个名为 AWS Data Wrangler 的新软件包。它将 Pandas 库的功能扩展到 AWS，以便与 Athena 和许多其他 AWS 服务轻松交互。

参考链接：

1. https://github.com/awslabs/aws-data-wrangler
2. https://github.com/awslabs/aws-data-wrangler/blob/master/tutorials/006%20-%20Amazon%20Athena.ipynb

注意： AWS Data Wrangler 库在 Glue Python shell 中默认不可用。要将其包含在 Python shell 中，请按照以下链接中的说明进行操作：

https://aws-data-wrangler.readthedocs.io/en/latest/install.html#aws-glue-python-shell-jobs

Answer 3

我有几个月用胶水，我用：

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

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

data_frame = spark.read.format("com.databricks.spark.csv")\
    .option("header","true")\
    .load(<CSVs THAT IS USING FOR ATHENA - STRING>)