Apache Arrow(PyArrow)を使って簡単かつ高速にParquetファイルに変換する

id  price   total   price_profit    total_profit    discount    visible name    created updated
1   20000   300000000   4.56    67.89   789012.34   True    QuietComfort 35 2019-06-14  2019-06-14 23:59:59

Traceback (most recent call last):
  File "/Users/cmuser/IdeaProjects/pyarrow/pyarrow_parquet.py", line 62, in <module>
    column_types
  File "/Users/cmuser/IdeaProjects/pyarrow/pyarrow_parquet.py", line 31, in get_pyarrow_table
    convert_options=convertoptions
  File "pyarrow/_csv.pyx", line 397, in pyarrow._csv.read_csv
  File "pyarrow/error.pxi", line 89, in pyarrow.lib.check_status
pyarrow.lib.ArrowNotImplementedError: CSV conversion to date32[day] is not supported

Traceback (most recent call last):
  File "/Users/cmuser/IdeaProjects/pyarrow/pyarrow_parquet.py", line 62, in <module>
    column_types
  File "/Users/cmuser/IdeaProjects/pyarrow/pyarrow_parquet.py", line 31, in get_pyarrow_table
    convert_options=convertoptions
  File "pyarrow/_csv.pyx", line 397, in pyarrow._csv.read_csv
  File "pyarrow/error.pxi", line 89, in pyarrow.lib.check_status
pyarrow.lib.ArrowNotImplementedError: CSV conversion to decimal(10, 2) is not supported

from pyarrow.csv import read_csv, ReadOptions, ParseOptions, ConvertOptions
from pyarrow import int16, int32, int64, float64, float32, bool_, date32, decimal128, timestamp, string, Table, parquet as pq
from boto3.session import Session


#
# Functions
#
def get_pyarrow_table(input_files: str, column_types: dict) -> Table:
    readoptions = ReadOptions(
        use_threads=True,  # 複数の読み取りスレッドの利用
        block_size=1000  # 読み取りブロック数
    )
    #
    convertoptions = ConvertOptions(
        check_utf8=True,  # 文字列カラムのUTF-8妥当性をチェック
        column_types=column_types,  # 列のデータ型を辞書型で渡す
        null_values=['']  # データ内のNULLを表す文字列
    )
    parseoptions = ParseOptions(
        delimiter='\t',  # タブ区切り指定
        double_quote=False,  # ダブルクオートで括らない
        escape_char='\'',  # エスケープ文字の指定
        header_rows=1  # 先頭がヘッダ行
    )
    pyarrow_table = read_csv(
        input_file=input_files,
        read_options=readoptions,
        parse_options=parseoptions,
        convert_options=convertoptions
    )
    return pyarrow_table


def tsv2parquet(input_file: str, output_file: str, column_types: dict):

    arrow_table = get_pyarrow_table(
        input_file,
        column_types
    )

    pq.write_table(
        arrow_table,
        output_file,
        compression='snappy',  # snappyで圧縮
        flavor=['spark'],  # spark互換の設定
    )


#
# main
#
def main():
    pyarrow_home = '/Users/cmuser/Desktop'
    tablename = 'sales'
    input_files = pyarrow_home + '/' + tablename + '.tsv'
    output_file = pyarrow_home + '/' + tablename + '.parquet'
    column_types = {
        'id': int16(),
        'price': int32(),
        'total': int64(),
        'price_profit': float32(),
        'total_profit': float64(),
        'discount': float64(),
        'visible': bool_(),
        'name': string(),
        'created': timestamp('s'),
        'updated': timestamp('s')
    }

    tsv2parquet(input_files, output_file, column_types)

    # Parquet -> S3
    s3bucket = 'cm-user'
    s3key = 'tmp/' + tablename + '/' + tablename + '.snappy.parquet'
    session = Session(profile_name='cm-user')
    s3 = session.resource('s3', region_name='ap-northeast-1')
    bucket = s3.Bucket(s3bucket)
    bucket.upload_file(output_file, s3key)


if __name__ == "__main__":
    main()

CREATE EXTERNAL TABLE `sales`(
  `id` smallint, 
  `price` int, 
  `total` bigint, 
  `price_profit` float, 
  `total_profit` double, 
  `discount` double, 
  `visible` boolean, 
  `name` string, 
  `created` timestamp, 
  `updated` timestamp)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://cm-user/tmp/sales/'
TBLPROPERTIES (
  'CrawlerSchemaDeserializerVersion'='1.0', 
  'CrawlerSchemaSerializerVersion'='1.0', 
  'UPDATED_BY_CRAWLER'='sales', 
  'averageRecordSize'='673', 
  'classification'='parquet', 
  'compressionType'='none', 
  'objectCount'='1', 
  'recordCount'='1', 
  'sizeKey'='2214', 
  'typeOfData'='file')

import pandas as pd
from pyarrow import int16, int32, int64, float64, float32, bool_, date32, decimal128, timestamp, string, Table, schema, parquet as pq
import decimal as dc
from boto3.session import Session


# テキストファイル読込み
def get_dataframe(input_file: str, pandas_dtype: dict, pandas_convert: dict, pandas_parse_dates: list) -> pd.DataFrame:
    dataframe = pd.read_csv(
        filepath_or_buffer=input_file,
        header=0,  # 0:ヘッダーあり
        encoding="utf-8",  # 文字コード
        sep='\t',  # セパレータ文字
        na_values='',  # NaNを置き換える文字列
        # compression='gzip', # gzip圧縮
        dtype=pandas_dtype,
        converters=pandas_convert,
        parse_dates=pandas_parse_dates,
        infer_datetime_format=True  # 日付データを読み込む速度を向上させる
    )
    return dataframe


# TSVファイルをParquetファイルに変換
def tsv2parquet(input_file: str, output_file: str, pandas_dtype: dict, pandas_convert: dict, pandas_parse_dates: list, pyarrow_schema: schema):
    dataframe = get_dataframe(
        input_file=input_file,
        pandas_dtype=pandas_dtype,
        pandas_convert=pandas_convert,
        pandas_parse_dates=pandas_parse_dates
    )
    pyarrow_table = Table.from_pandas(
        dataframe,
        pyarrow_schema,
        preserve_index=False
    )
    pq.write_table(
        pyarrow_table,
        output_file,
        compression='snappy',  # 圧縮タイプ指定
        flavor=['spark'],  # 互換性のためsparkを設定
        use_deprecated_int96_timestamps=True  # int96のParquet形式で書き込み
    )


#
# main
#
def main():
    pyarrow_home = '/Users/cmsuer/Desktop'
    tablename = 'sales'
    input_files = pyarrow_home + '/' + tablename + '.tsv'
    output_file = pyarrow_home + '/' + tablename + '.parquet'

    # Pandasに読み込むデータ定義、
    pandas_dtype = {
        'id': float,  # NaN対策（int -> float）
        'price': float,  # NaN対策（int -> float）
        'total': float,  # NaN対策（int -> float）
        'price_profit': float,
        'total_profit': float,
        # 'discount': decimal(10, 2),
        'visible': bool,
        'name': str,
        # 'created': timestamp,
        # 'updated': timestamp
    }

    # Pandasに読み込んだ型変換
    pandas_convert = {'discount': dc.Decimal}

    # 日時型のカラムの指定
    pandas_parse_dates = ['created', 'updated']

    # PandasからPyArrowのテーブルに変換するテータ定義
    pyarrow_schema = schema({
        'id': int16(),
        'price': int32(),
        'total': int64(),
        'price_profit': float32(),
        'total_profit': float64(),
        'discount': decimal128(10, 2),
        'visible': bool_(),
        'name': string(),
        'created': timestamp('s'),
        'updated': timestamp('s')
    })

    # TSVファイルをParquetファイルに変換
    tsv2parquet(input_files, output_file, pandas_dtype, pandas_convert, pandas_parse_dates, pyarrow_schema)

    # ParquetファイルをS3にアップロード
    s3bucket = 'cm-user'
    s3key = 'tmp/' + tablename + '/' + tablename + '.snappy.parquet'
    session = Session(profile_name='cmuser')
    s3 = session.resource('s3', region_name='ap-northeast-1')
    bucket = s3.Bucket(s3bucket)
    bucket.upload_file(output_file, s3key)


if __name__ == "__main__":
    main()

CREATE EXTERNAL TABLE `sales`(
  `id` smallint, 
  `price` int, 
  `total` bigint, 
  `price_profit` float, 
  `total_profit` double, 
  `discount` decimal(10,2), 
  `visible` boolean, 
  `name` string, 
  `created` timestamp, 
  `updated` timestamp)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://cm-user/tmp/sales/'
TBLPROPERTIES (
  'CrawlerSchemaDeserializerVersion'='1.0', 
  'CrawlerSchemaSerializerVersion'='1.0', 
  'UPDATED_BY_CRAWLER'='sales', 
  'averageRecordSize'='648', 
  'classification'='parquet', 
  'compressionType'='none', 
  'objectCount'='1', 
  'recordCount'='1', 
  'sizeKey'='3424', 
  'typeOfData'='file')