自定义表数据导入功能

一、概述

    在一些情况下，您需要运用到的数据可能无法以user或event的形式来表示，比如一些映射关系表, 或者一些外部数据，如果需要运用这部分数据，则需要通过data_transfer命令导入自定义数据到 TA 系统中，结合事件表和用户表关联使用。
    目前支持以下三种导入数据源：

• mysql: 远程 mysql 数据库

• txtfile: 本地文件

二、使用说明

2.1 命令说明

    数据导入的命令如下：

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

2.2 命令参数说明

2.2.1 -conf

    传入的参数为导入表的配置文件路径，每一个表即为一个配置文件，支持多个表同时导入，支持通配符的方式，例如：/data/config/*或者 ./config/*.json

2.2.1 --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 三种导入数据源，后续会增加更多数据源的支持
  • 类型: string
  • 必选：是
  • 默认值：无

• parameter

  • 描述： 不同数据源的具体配置，在三、导入数据源配置
  • 类型: jsonObject
  • 必选：是
  • 默认值：无

2.3.4 target 参数具体说明

• appid

  • 描述： 导入表对应的项目 appid，在 TA 系统后台可以查到得到
  • 类型: string
  • 必选：是
  • 默认值：无

• table

  • 描述： 导入到 TA 系统中的表名， 注意：表名全局不能重复，建议基于不同项目增加可以区分的前缀或后缀
  • 类型: string
  • 必选：是
  • 默认值：无

• table_desc

  • 描述： 导入表的注释，建议导入时配置该参数，方便后续查询表时明确表含义
  • 类型: string
  • 必选：否
  • 默认值：空

• partition_val

  • 描述： 导入的分区值，TA 系统导入的自定义表默认都会带上分区字段 $pt ，所以在导入时必须指定导入的分区值，一般可以设定为导入的数据日期，例如: 20180701, 同时也支持时间宏替换，例如：@[{yyyyMMdd}-{1day}]，在 2.1 节将会介绍具体使用方法
  • 类型: string
  • 必选：是
  • 默认值：无

• column

  • 描述： 定义导入到 TA 系统中的表字段定义，包含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-31
  • @[{yyyyMMddHH}+{2hour}] 替换为 2018070117
  • @[{yyyyMMddHHmm00}-{10minute}] 替换为 20180701150300

三、导入数据源配置

    本节将会介绍不同数据源的参数配置，目前支持：txtfile, mysql, ftp 三种导入数据源，根据数据源的不同，需要您调整 source 的参数

3.1 mysql 数据源

    数据源通过 JDBC 连接器连接到远程的 mysql 数据库，并根据用户配置的信息生成查询 SELECT SQL 语句，然后发送到远程 mysql 数据库，并将该 SQL 执行返回结果导入到 TA 系统的表内

3.1.1 配置样例

• mysql 整表导入到 TA 系统的配置样例：

{
  "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 导入到 TA 系统的配置样例:

{
  "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 连接即可。
  • 类型: 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 数据源是读本地服务器上的文件，导入到 TA 的系统表内，目前 txtfile 的使用限制及特性如下：

1. 支持且仅支持读取 TXT 的文件，且要求 TXT 中 schema 为一张二维表
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目录下所有的文件。目前只支持*号作为文件通配符。 特别需要注意的是，导入工具会将一个作业下同步的所有 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提供nullFormat定义哪些字符串可以表示为null。 例如如果用户配置: nullFormat:"\N"，那么如果源头数据是"\N"，ta-tool 视作null字段。
  • 类型: string
  • 必选：否
  • 默认值：\N