Redshiftに対してCopyコマンドを発行してみた(JSON取り込み) | Luigi Advent Calendar 2016 #19

{
  "entries": [
    {
      "url": "s3://cm-kajiwara-redshift-load/100_199/000102.json",
      "mandatory": true
    },
(省略)
    {
      "url": "s3://cm-kajiwara-redshift-load/100_199/000198.json",
      "mandatory": true
    }
  ]
}

$ cat ./only3manifest| jq '.entries[]' > strip3Manifest.json
$ cat ./strip3Manifest.json
{
    "url": "s3://cm-kajiwara-redshift-load/100_199/000102.json",
    "mandatory": true
}
(省略)
{
    "url": "s3://cm-kajiwara-redshift-load/100_199/000198.json",
    "mandatory": true
}

{
    "jsonpaths": [
        "$['url']"
    ]
}

create table manifest (url varchar(1024));

import luigi
import luigi.contrib.redshift

class copyJsonPath(luigi.contrib.redshift.S3CopyJSONToTable):

    s3FilePath = luigi.Parameter()
    param = {}
    param["host"] = "[Redshift Host]"
    param["database"] = "[Redshift database]"
    param["password"] = "[Redshift Password]"
    param["user"] = "[Redshift User]"
    param["table"] = "顧客情報"
    param["s3_load_path"] = "s3://cm-kajiwara-redshift-load/{}
    param["jsonpath"] = "s3://cm-kajiwara-redshift-load/jsonpath.json"
    param["copy_options"] = "region AS 'ap-northeast-1' DELIMITER AS '\t' TIMEFORMAT AS 'auto' "
    param["copy_json_options"] = ""
    param["aws_access_key_id"] = "["aws_access_key_id"]"
    param["aws_secret_access_key"] = "["aws_secret_access_key"]"

    @property
    def host(self):
        return self.param["host"]

    @property
    def database(self):
        return self.param["database"]

    @property
    def user(self):
        return self.param["user"]

    @property
    def password(self):
        return self.param["password"]

    @property
    def table(self):
        return self.param["table"]

    def s3_load_path(self):
        return self.param["s3_load_path"].format(self.s3FilePath)

    @property
    def jsonpath(self):
        return self.param["jsonpath"]

    @property
    def aws_access_key_id(self):
        return self.param["aws_access_key_id"]

    @property
    def aws_secret_access_key(self):
        return self.param["aws_secret_access_key"]

    @property
    def copy_json_options(self):
        return self.param["copy_json_options"]

    @property
    def copy_options(self):
        return self.param["copy_options"]

$ python ./useRedshift.py --local-scheduler copyJsonPath --s3FilePath strip3Manifest.json
DEBUG: Checking if copyJsonPath(s3FilePath=strip3Manifest.json) is complete
INFO: Informed scheduler that task   copyJsonPath_strip3Manifest_j_8bec9a68a0   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 36672] Worker Worker(salt=996099161, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=36672) running   copyJsonPath(s3FilePath=strip3Manifest.json)
INFO: Inserting file: s3://cm-kajiwara-redshift-load/strip3Manifest.json
INFO: Executing post copy queries
INFO: [pid 36672] Worker Worker(salt=996099161, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=36672) done      copyJsonPath(s3FilePath=strip3Manifest.json)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   copyJsonPath_strip3Manifest_j_8bec9a68a0   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=996099161, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=36672) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 1 tasks of which:
* 1 ran successfully:
    - 1 copyJsonPath(s3FilePath=strip3Manifest.json)

This progress looks :) because there were no failed tasks or missing external dependencies

===== Luigi Execution Summary =====