将 Athena 中的数据加载到 AWS 中 Cloud9/Lambda 中的 Pandas 数据帧以进行 ETL答案

【问题标题】：Load data from Athena into Pandas dataframe in Cloud9/Lambda in AWS for ETL将 Athena 中的数据加载到 AWS 中 Cloud9/Lambda 中的 Pandas 数据帧以进行 ETL
【发布时间】：2019-04-08 11:30:18
【问题描述】：

我正在 AWS 中构建数据湖。源数据作为 CDC 导入 S3。我需要找到一种方法来合并它们，以便拥有一个包含最新版本信息的表格。

最初我想使用 Glue 进行 ETL 开发，但编辑器似乎相当笨重。此外，数据量并没有太大以至于需要火花。 Pandas 也将发挥作用，并且在组织中拥有更广泛的知识库。

所以我使用 Glue 来抓取导入，现在我想要在 Cloud9 中开发我的聚合以稍后迁移到 Lambda 函数的 Athena 表。

问题是我无法将 Athena 数据放入数据框中。

我已经尝试了 boto3 的 start_query_execution 函数，但它不返回数据，而只是将它写入我不想要的 S3。它还作为 QueryExecutionId 返回，我已将其传递给另一个名为 get_query_results 的 boto 函数。似乎有响应，但我在如何将数据传递到数据框（是 JSON 还是 dict？）上苦苦挣扎。

#python 3.6
import pandas as pd
import numpy as np
import boto3
import time

#https://dev.classmethod.jp/cloud/run-amazon-athenas-query-with-aws-lambda/

#athena constant
DATABASE = 'myDatabase'
TABLE = 'myTable'

#output
S3_OUTPUT = 's3://myBucket/myPath/'

client = boto3.client('athena')

response = client.start_query_execution(
        QueryString='select * from myTable limit 100',
        QueryExecutionContext={
            'Database': DATABASE
        },
        ResultConfiguration={
            'OutputLocation': S3_OUTPUT,

        }
)

print(response["QueryExecutionId"])

time.sleep(50)

data = client.get_query_results(
    QueryExecutionId=response["QueryExecutionId"]
)

dataDf = pd.read_json(data["ResultSet"])
print(dataDf.head())

【问题讨论】：

标签： pandas aws-lambda etl amazon-athena aws-cloud9

【解决方案1】：

这对我有用。下载文件而不是使用 JSON 响应。

import os
import boto3

s3 = boto3.client('s3')
bucket = 'myBucket'
key = 'myPath'

data_file_name = f'{response["QueryExecutionId"]}.csv'
object = os.path.join(key, data_file_name)
s3.download_file(bucket, object, data_file_name)
df = pd.read_csv(data_file_name)

【讨论】：