AthenaのMSCK REPAIR TABLEを試してみた

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([
    StructField('jis_code', StringType(), True, {'comment': "全国地方公共団体コード"}),
    StructField('postal5', StringType(), True, {'comment': "（旧）郵便番号（5桁）"}),
    StructField('postal7', StringType(), True, {'comment': "郵便番号（7桁）"}),
    StructField('pref_kana', StringType(), True, {'comment': "都道府県名カナ"}),
    StructField('city_kana', StringType(), True, {'comment': "市区町村名カナ"}),
    StructField('address_kana', StringType(), True, {'comment': "町域名カナ"}),
    StructField('pref', StringType(), True, {'comment': "都道府県名"}),
    StructField('city', StringType(), True, {'comment': "市区町村"}),
    StructField('address', StringType(), True, {'comment': "町域名"}),
    StructField('flag1', StringType(), True, {'comment': "一町域が二以上の郵便番号で表される場合の表示（「1」は該当、「0」は該当せず）"}),
    StructField('flag2', StringType(), True, {'comment': "小字毎に番地が起番されている町域の表示（「1」は該当、「0」は該当せず）"}),
    StructField('flag3', StringType(), True, {'comment': "丁目を有する町域の場合の表示　（「1」は該当、「0」は該当せず）"}),
    StructField('flag4', StringType(), True, {'comment': "一つの郵便番号で二以上の町域を表す場合の表示（「1」は該当、「0」は該当せず）"}),
    StructField('flag5', StringType(), True, {'comment': "更新の表示（「0」は変更なし、「1」は変更あり、「2」廃止（廃止データのみ使用））"}),
    StructField('flag6', StringType(), True, {'comment': "変更理由　（「0」は変更なし、「1」市政・区政・町政・分区・政令指定都市施行、「2」住居表示の実施、「3」区画整理、「4」郵便区調整等、「5」訂正、「6」廃止（廃止データのみ使用））"})
])

def main():
    spark = SparkSession.builder \
        .config('hive.metastore.client.factory.class', 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory') \
        .config('spark.sql.sources.partitionOverwriteMode', 'dynamic') \
        .enableHiveSupport() \
        .getOrCreate()
    # s3から東京都のcsvデータを入力
    df = spark.read \
        .format('csv') \
        .option('header','false') \
        .option('encoding','Windows-31J') \
        .schema(schema) \
        .load('{_s3bucket_input}/tokyo/')
    # s3にparquet形式のデータを出力してGlueカタログにテーブルメタデータを登録
    spark.catalog.setCurrentDatabase('test')
    df.write.option('path','{_s3bucket_output}/zipcode') \
        .option('compression','snappy') \
        .mode('overwrite') \
        .partitionBy('pref') \
        .format('parquet') \
        .saveAsTable('zipcode')

if __name__ == '__main__':
    main()

def main():
    spark = SparkSession.builder \
        .config('hive.metastore.client.factory.class', 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory') \
        .config('spark.sql.sources.partitionOverwriteMode', 'dynamic') \
        .enableHiveSupport() \
        .getOrCreate()
    df = spark.read \
        .format('csv') \
        .option('header','false') \
        .option('encoding','Windows-31J') \
        .schema(schema) \
        .load('{_s3bucket_input}/chiba/')
    df.write.parquet('{_s3bucket_output}/zipcode', mode='overwrite', partitionBy='pref', compression='snappy')

def main():
    spark = SparkSession.builder \
        .config('hive.metastore.client.factory.class', 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory') \
        .enableHiveSupport() \
        .getOrCreate()
    spark.sql('MSCK REPAIR TABLE test.zipcode')