カスタムテーブルデータのインポート機能

最終更新 2026/10/03

1. 概要

場合によっては、使用したいデータをuserやeventの形式で表現できないことがあります。たとえば、マッピング関係テーブルや外部データなどです。このようなデータを使用する必要がある場合は、data_transferコマンドでカスタムデータをAEシステムにインポートし、イベントテーブルやユーザーテーブルと関連付けて使用します。

現在、次の2種類のインポートデータソースに対応しています：

• mysql: リモートのmysqlデータベース
• txtfile: ローカルファイル

2. 使用方法

2.1 コマンドの説明

データインポートのコマンドは次のとおりです：

ta-tool data_transfer -conf <config files> [--date xxx]

2.2 コマンドパラメータの説明

2.2.1 -conf

渡すパラメータは、インポートするテーブルの設定ファイルのパスです。1つのテーブルが1つの設定ファイルに対応します。複数のテーブルの同時インポートに対応し、ワイルドカードも使用できます。例：/data/config/*または./config/*.json

2.2.2 --date

オプションパラメータ--date：任意。データの日付を表すパラメータで、時間マクロはこの基準時刻に基づいて置換されます。省略可能で、省略した場合はデフォルトで現在の日付が使用されます。形式はYYYY-MM-DDです。時間マクロの具体的な使用方法は、時間マクロの使用方法を参照してください

2.3 設定ファイルの説明

2.3.1 単一テーブルの設定ファイルの例は次のとおりです：

{
  "parallel_num": 2,
  "source": {
    "type": "txtfile",
    "parameter": {
      "path": ["/data/home/ta/importer_test/data/*"],
      "encoding": "UTF-8",
      "column": ["*"],
      "fieldDelimiter": "\t"
    }
  },
  "target": {
    "appid": "test-appid",
    "table": "test_table",
    "table_desc": "インポートテスト用テーブル",
    "partition_value": "@[{yyyyMMdd}-{1day}]",
    "column": [
      {
        "name": "col1",
        "type": "timestamp",
        "comment": "タイムスタンプ"
      },
      {
        "name": "col2",
        "type": "varchar"
      }
    ]
  }
}

2.3.2 外側のパラメータの説明

• parallel_num

  • 説明：インポートの並列スレッド数。インポートの速度を制御します
  • タイプ：int
  • 必須：はい
  • デフォルト値：なし

• source

  • 説明：インポートデータソースの具体的なパラメータ設定
  • タイプ：jsonObject
  • 必須：はい
  • デフォルト値：なし

• target

  • 説明：インポート先テーブルの具体的なパラメータ設定
  • タイプ：jsonObject
  • 必須：はい
  • デフォルト値：なし

2.3.3 sourceパラメータの詳細説明

• type

  • 説明：インポートデータソースのタイプ。現在、インポートツールはtxtfile, mysql, ftpの3種類のインポートデータソースに対応しており、今後さらに多くのデータソースに対応する予定です
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• parameter

  • 説明：データソースごとの具体的な設定は、3. インポートデータソースの設定を参照してください
  • タイプ：jsonObject
  • 必須：はい
  • デフォルト値：なし

2.3.4 targetパラメータの詳細説明

• appid

  • 説明：インポートするテーブルに対応するプロジェクトのappid。AEシステムの管理画面で確認できます
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• table

  • 説明：AEシステムにインポートするテーブル名。注意：テーブル名はグローバルで重複できません。プロジェクトごとに区別できるプレフィックスまたはサフィックスを付けることをお勧めします
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• table_desc

  • 説明：インポートするテーブルのコメント。後でテーブルを照会する際にテーブルの意味がわかるよう、インポート時にこのパラメータを設定することをお勧めします
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：空

• partition_value

  • 説明：インポートのパーティション値。AEシステムにインポートしたカスタムテーブルにはデフォルトでパーティションフィールド$ptが付くため、インポート時にはパーティション値を必ず指定する必要があります。通常はインポートするデータの日付を設定します。例：20180701。時間マクロによる置換にも対応しています。例：@[{yyyyMMdd}-{1day}]。具体的な使用方法は2.4節で説明します
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• column

  • 説明：AEシステムにインポートするテーブルのフィールド定義です。name, type, commentの3つのプロパティ値を含み、そのうちnameとtypeは必須フィールドです。例は次のとおりです：

[
  {
    "name": "col1",
    "type": "timestamp",
    "comment": "タイムスタンプ"
  },
  {
    "name": "col2",
    "type": "varchar"
  }
]

source側がmysqlで、テーブル全体をインポートする場合（つまりcolumnフィールドが["*"]の場合）、targetではcolumnパラメータを省略でき、インポートツールはmysqlのテーブル構造に従います。それ以外の場合、このフィールドは必須です

• タイプ：jsonArray
• 必須：いいえ
• デフォルト値：mysql source側のテーブルのschema定義

2.4 時間マクロの使用方法

設定ファイル内では、時間マクロを使って時間パラメータを置換できます。ta-toolツールはインポートの開始時刻を基準に、時間マクロのパラメータに基づいて時間のオフセットを計算し、設定ファイル内の時間マクロを置換します。対応している時間マクロの形式：@[{yyyyMMdd}], @[{yyyyMMdd}-{nday}], @[{yyyyMMdd}+{nday}]など

• yyyyMMddは、JavaのdateFormatで解析できる任意の日付形式に置き換えられます。例：yyyy-MM-dd HH:mm:ss.SSS, yyyyMMddHH000000

• nは任意の整数で、時間のオフセット値を表します

• dayは時間のオフセット単位を表し、次のいずれかを指定できます: day, hour, minute, week, month

• 例：現在時刻が2018-07-01 15:13:23.234であるとします

  • @[{yyyyMMdd}]は20180701に置換されます
  • @[{yyyy-MM-dd}-{1day}]は2018-06-30に置換されます
  • @[{yyyyMMddHH}+{2hour}]は2018070117に置換されます
  • @[{yyyyMMddHHmm00}-{10minute}]は20180701150300に置換されます

3. インポートデータソースの設定

このセクションでは、データソースごとのパラメータ設定について説明します。現在、txtfile, mysql, ftpの3種類のインポートデータソースに対応しています。データソースに応じて、sourceのパラメータを調整してください

3.1 mysqlデータソース

データソースはJDBCコネクタ経由でリモートのmysqlデータベースに接続し、ユーザーが設定した情報に基づいてSELECTクエリのSQL文を生成してリモートのmysqlデータベースに送信します。そして、そのSQLの実行結果をAEシステムのテーブルにインポートします

3.1.1 設定例

• mysqlのテーブル全体をAEシステムにインポートする設定例：

{
  "parallel_num": 2,
  "source": {
    "type": "mysql",
    "parameter": {
      "username": "test",
      "password": "test",
      "column": ["*"],
      "connection": [
        {
          "table": ["test_table"],
          "jdbcUrl": ["jdbc:mysql://mysql-ip:3306/testDb"]
        }
      ]
    }
  },
  "target": {
    "appid": "test-appid",
    "table": "test_table_abc",
    "table_desc": "mysqlテスト用テーブル",
    "partition_value": "@[{yyyy-MM-dd}-{1day}]"
  }
}

• カスタムSQLでAEシステムにインポートする設定例:

{
  "parallel_num": 1,
  "source": {
    "type": "mysql",
    "parameter": {
      "username": "test",
      "password": "test",
      "connection": [
        {
          "querySql": [
            "select db_id,log_time from test_table where log_time>='@[{yyyy-MM-dd 00:00:00}-{1day}]' and log_time<'@[{yyyy-MM-dd 00:00:00}]'"
          ],
          "jdbcUrl": ["jdbc:mysql://mysql-ip:3306/testDb"]
        }
      ]
    }
  },
  "target": {
    "appid": "test-appid",
    "table": "test_table_abc",
    "table_desc": "mysqlテスト用テーブル",
    "partition_value": "@[{yyyy-MM-dd}-{1day}]",
    "column": [
      {
        "name": "db_id",
        "type": "bigint",
        "comment": "dbの連番"
      },
      {
        "name": "log_time",
        "type": "timestamp",
        "comment": "タイムスタンプ"
      }
    ]
  }
}

3.1.2 parameterパラメータの説明

• jdbcUrl

  • 説明：接続先データベースへのJDBC接続情報をJSONの配列で記述します。注意：jdbcUrlはconnection設定ユニットに含める必要があります。通常、JSON配列にはJDBC接続を1つ記入すれば十分です。
  • タイプ：jsonArray
  • 必須：はい
  • デフォルト値：なし

• username

  • 説明：データソースのユーザー名
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• password

  • 説明：データソースで指定したユーザー名のパスワード
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• table

  • 説明：同期対象として選択したテーブル。JSONの配列で記述するため、複数のテーブルを同時に抽出できます。複数のテーブルを設定する場合は、それらが同じschema構造であることをユーザー自身が保証する必要があります。MysqlReaderは、テーブルが同一の論理テーブルかどうかをチェックしません。注意：tableはconnection設定ユニットに含める必要があります。
  • タイプ：jsonArray
  • 必須：はい
  • デフォルト値：なし

• column

  • 説明：設定したテーブルのうち同期が必要な列名の集合で、JSONの配列でフィールド情報を記述します。*を使うと、デフォルトですべての列を使用する設定になります。例：["*"]。
  • タイプ：jsonArray
  • 必須：はい
  • デフォルト値：なし

• where

  • 説明：フィルター条件。指定したcolumn、table、where条件に基づいてSQLを組み立て、そのSQLに基づいてデータを抽出します。実際の業務シナリオでは、前日のデータを選んで同期することが多く、where条件をlog_time>='@[{yyyy-MM-dd 00:00:00}-{1day}]' and log_time<'@[{yyyy-MM-dd 00:00:00}]'と指定できます。注意：where条件にlimit 10を指定することはできません。limitはSQLの有効なwhere句ではありません。where条件を使うと、業務の増分同期を効率的に行えます。where文を記入しない場合、インポートツールは全量データの同期とみなします。
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：なし

• querySql

  • 説明：業務シナリオによっては、whereの設定項目だけではフィルター条件を記述しきれない場合があります。その場合は、この設定パラメータでフィルター用のSQLをカスタマイズできます。この項目を設定すると、インポートツールはtable、columnなどの設定パラメータを無視し、この設定項目の内容をそのまま使ってデータをフィルタリングします。たとえば、複数テーブルをjoinしてからデータを同期する場合: select a,b from table_a join table_b on table_a.id = table_b.id。querySqlを設定した場合、インポートツールはtable、column、where条件の設定を無視します。querySqlの優先度は、table、column、whereオプションより高くなります。
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：なし

3.2 txtfileデータソース

txtfileデータソースは、ローカルサーバー上のファイルを読み取り、AEのシステムテーブルにインポートします。現在のtxtfileの使用制限と特徴は次のとおりです：

1. TXTファイルの読み取りのみに対応しており、TXT内のschemaは2次元テーブルである必要があります
2. CSV形式に類するファイルとカスタム区切り文字に対応しています
3. 複数タイプのデータの読み取り(stringで表現)、列の切り出し、列定数に対応しています
4. 再帰的な読み取りとファイル名によるフィルタリングに対応しています
5. テキスト圧縮に対応しています。現在対応している圧縮形式はzip、gzip、bzip2です

3.2.1 設定例

{
  "parallel_num": 5,
  "source": {
    "type": "txtfile",
    "parameter": {
      "path": ["/home/ftp/data/testData/*"],
      "column": [
        {
          "index": 0,
          "type": "long"
        },
        {
          "index": 1,
          "type": "string"
        }
      ],
      "encoding": "UTF-8",
      "fieldDelimiter": "\t"
    }
  },
  "target": {
    "appid": "test-appid",
    "table": "test_table_abc",
    "table_desc": "mysqlテスト用テーブル",
    "partition_value": "@[{yyyy-MM-dd}-{1day}]",
    "column": [
      {
        "name": "db_id",
        "type": "bigint",
        "comment": "dbの連番"
      },
      {
        "name": "log_time",
        "type": "timestamp",
        "comment": "タイムスタンプ"
      }
    ]
  }
}

3.2.2 parameterパラメータの説明

• path

  • 説明：ローカルファイルシステムのパス情報です。ここには複数のパスを記入できます。ワイルドカードを指定すると、インポートツールは複数のファイル情報を走査しようとします。例: /data/*を指定すると、/dataディレクトリ配下のすべてのファイルを読み取ります。現在、ファイルのワイルドカードとして使用できるのは*のみです。特に注意が必要なのは、インポートツールが1つのジョブで同期するすべてのText Fileを同一のデータテーブルとみなす点です。すべてのFileが同じschema情報に適合することを、ユーザー自身が保証する必要があります。読み取るファイルはCSV形式に類するものである必要があります。
  • タイプ：string
  • 必須：はい
  • デフォルト値：なし

• column

  • 説明：読み取るフィールドのリストです。typeはソースデータのタイプを指定し、indexは現在の列がテキストの何列目(0から開始)に由来するかを指定します。valueは現在のタイプを定数として指定し、ソースファイルからデータを読み取らずに、valueの値に基づいて対応する列を自動生成します。

デフォルトでは、すべてのデータをstringタイプとして読み取れます。設定は次のとおりです：

 "column": ["*"]

Columnのフィールド情報を指定することもできます。設定は次のとおりです：

({
  "type": "long",
  "index": 0
},
{
  "type": "string",
  "value": "2018-07-01 00:00:00"
})

Column情報を指定する場合、typeは必須で、index/valueのいずれか一方を選択する必要があります。

typeの値の範囲：long、double、string、boolean

• タイプ：jsonArray
• 必須：はい
• デフォルト値：なし

• fieldDelimiter

  • 説明：読み取り時のフィールド区切り文字
  • タイプ：string
  • 必須：はい
  • デフォルト値：,

• compress

  • 説明：テキストの圧縮タイプです。デフォルトでは空欄で、圧縮なしを意味します。対応している圧縮タイプはzip、gzip、bzip2です。
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：圧縮なし

• encoding

  • 説明：読み取るファイルのエンコーディング設定です。
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：utf-8

• skipHeader

  • 説明：CSV形式に類するファイルでは、先頭行がヘッダー（タイトル）になっていてスキップが必要な場合があります。デフォルトではスキップしません。
  • タイプ：boolean
  • 必須：いいえ
  • デフォルト値：false

• nullFormat

  • 説明：テキストファイルでは標準の文字列でnull(ヌルポインタ)を定義できないため、ta-toolには、どの文字列をnullとして扱うかを定義するnullFormatが用意されています。たとえば、ユーザーがnullFormat:"\N"と設定した場合、ソースデータが"\N"であれば、ta-toolはそれをnullフィールドとみなします。
  • タイプ：string
  • 必須：いいえ
  • デフォルト値：\N