Конвертируйте файл из csv в паркет на S3 с помощью aws boto

Question

Я написал скрипт, который выполнял бы запрос в Афине и загружал файл результатов в указанном месте aws boto S3.

import boto3
def run_query(query, database, s3_output):
    client = boto3.client('athena', region_name='my-region')
    response = client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={
            'Database': database
            },
        ResultConfiguration={
            'OutputLocation': s3_output,
            }
        )
    print('Execution ID: ' + response['QueryExecutionId'])
    return response

query = """select ..."""

database = 'db_name'

path_template = 's3://bucket_name/path/version={}'

current_time =  str(datetime.datetime.now())

result = run_query(query, database, path_template.format(current_time))

Это работает, но проблема в том, что у меня есть файл csv какуказанное место. Но мне не нужен CSV-файл. Мне нужен паркетный файл.

Единственный способ получить то, что я хочу, - это загрузить файл, преобразовать его с помощью panda в паркет, чтобы выгрузить его. Раздражает, что я не могу просто конвертировать напрямую без загрузки файла.

Кто-нибудь может предложить другой способ? Я не хочу использовать CTAS.

Answer 1

В комментариях к ответу @Istvan @Джон Ротенштейн предложил мне сделать это с помощью запроса INSERT INTO.

https://aws.amazon.com/fr/about-aws/whats-new/2019/09/amazon-athena-adds-support-inserting-data-into-table-results-of-select-query/

https://docs.aws.amazon.com/fr_fr/athena/latest/ug/insert-into.html

Answer 2

Вам необходимо использовать CTAS:

CREATE TABLE db.table_name
WITH (
    external_location = 's3://yourbucket/path/table_name',
    format = 'PARQUET',
    parquet_compression = 'GZIP',
    partitioned_by = ARRAY['dt']
)
AS
SELECT
    ...
;

Таким образом, результат выбора будет сохранен как паркет.

https://docs.aws.amazon.com/athena/latest/ug/ctas-examples.html

https://docs.aws.amazon.com/athena/latest/ug/ctas.html

Использование запросов CTAS для:

• Создание таблиц из результатов запроса за один шаг без повторного запроса необработанных наборов данных. Это облегчает работу с необработанными наборами данных.
• Преобразование результатов запроса в другие форматы хранения, такие как Parquet и ORC. Это повышает производительность запросов и снижает стоимость запросов в Афине. Для получения информации см. Форматы хранения столбцов.
• Создание копий существующих таблиц, содержащих только те данные, которые вам нужны.

ОБНОВЛЕНИЕ (2019.10):

AWS только что выпустил INSERT INTO для Афины.

https://docs.aws.amazon.com/en_pv/athena/latest/ug/insert-into.html

Вставляет новые строки в таблицу назначения на основе оператора запроса SELECT, который выполняется на исходной таблице, или на основе набора VALUES, предоставляемых как частьзаявление. Если исходная таблица основана на базовых данных в одном формате, например CSV или JSON, а таблица назначения основана на другом формате, например, Parquet или ORC, вы можете использовать запросы INSERT INTO для преобразования выбранных данных в формат таблицы назначения. .

Существуют некоторые ограничения:

• INSERT INTO не поддерживается для таблиц с интервалами. Дополнительную информацию см. В разделе Bucketing vs Partitioning.
• При выполнении запроса INSERT для таблицы с базовыми данными, зашифрованными в Amazon S3, выходные файлы, которые записывает запрос INSERT, по умолчанию не шифруются. Мы рекомендуем вам шифровать результаты запроса INSERT, если вы вставляете в таблицы зашифрованные данные. Дополнительную информацию о шифровании результатов запросов с помощью консоли см. В разделе Шифрование результатов запросов, хранящихся в Amazon S3. Чтобы включить шифрование с помощью интерфейса командной строки AWS или API Athena, используйте свойства EncryptionConfiguration действия StartQueryExecution, чтобы указать параметры шифрования Amazon S3 в соответствии с вашими требованиями.