S3 Multipart Upload와 pyarrow로 고정된 스키마를 가지는 Parquet 파일 생성

import pyarrow as pa

schema: pa.Schema = pa.schema(
    [
        pa.Field("id", pa.int64()),
        pa.Field("name", pa.string()),
        pa.Field("created_dttm", pa.timestamp('us')), # parquet에서 nanosecond timestamp는 deprecate되었음. https://issues.apache.org/jira/browse/ARROW-1957
        pa.Field("cost", pa.decimal128(18,10)),
        # 이외의 data type은 https://arrow.apache.org/docs/python/api/datatypes.html
    ]
)

t: pa.Table = pa.Table.from_pylist(
    [
        {"id": 1, "name": "name1", "created_dttm": datetime.now(), "cost": Decimal(10.0)},
        ...
    ],
)

from pyarrow import fs, parquet as pq

# https://arrow.apache.org/docs/python/generated/pyarrow.fs.S3FileSystem.html#pyarrow.fs.S3FileSystem
file_system: fs.FileSystem = fs.S3FileSystem(
    access_key=access_key,
    secret_key=secret_key,
    region=region,
    # background_writes=True, # True가 default value인데 non-blocking으로 write를 하게된다.
)

# https://arrow.apache.org/docs/python/generated/pyarrow.parquet.ParquetWriter.html
parquet_writer: pq.ParquetWriter = pq.ParquetWriter(
    where=f's3://{bucket}/{prefix}/{object_name}.parquet'
    schema=schema, # 위에서 생성한 테이블 스키마,
    filesystem=file_system,
    allow_truncated_timestamp=True, # timestamp에서 nanoseconde등의 data loss가 생길때 무시하는 옵션
)

parquet_writer.write_table(t1) # 매번 write_table할때마다 
parquet_writer.write_table(t2)
...
parquet_writer.write_table(tn)

parquet_writer.close() # with as statement로도 사용가능하다.

records: Tuple[Tuple[Any], ...] = (
    (1, "name1", '2022-05-01T00:00:00+00:00', '10.0'),
    (2, "name2", '2022-05-01T00:00:00+00:00', '11.0'),
    ...
)
column_wise: List[List[Any]] = []
for record in records:
    for i, data in enumerate(record):
        column_wise[i].append(data)

t: pa.Table = pa.Table.from_arrays(
    arrays=[pa.array(column_list) for column_list in columns],
    schema=schema,
)

from decimal import Decimal

records: Tuple[Tuple[Any], ...] = (
    (1, "name1", '2022-05-01T00:00:00+00:00', ''),
    (2, "name2", '2022-05-01T00:00:00+00:00', None),
    ...
)
column_wise: List[List[Any]] = []
for record in records:
    for i, data in enumerate(record):
        if pa.types.is_decimal(schema.types[i]): # pa.Schema에서 i번째 컬럼이 Decimal type인지 확인
            column_wise[i].append(Decimal(data) if data is not None and data != '' else None) # Decimal(None)은 에러가 난다.
        else:
            column_wise[i].append(data)

from mypy_boto3_s3 import S3Client # 이 라이브러리를 활용하면 boto에 type을 씌워서 쓸 수 있다
from mypy_boto3_s3.type_defs import ListMultipartUploadsOutputTypeDef

s3: S3Client = boto3.client('s3')

res: ListMultipartUploadsOutputTypeDef = s3.list_multipart_uploads(bucket=bucket, prefix=prefix)
upload: MultipartUploadTypeDef
for upload in res['Uploads']:
    s3.abort_multipart_upload(bucket=bucket, key=upload['Key'], UploadId=upload['UploadId'])