DataX FtpWriter 说明 1 快速介绍 FtpWriter提供了向远程FTP文件写入CSV格式的一个或者多个文件，在底层实现上，FtpWriter将DataX传输协议下的数据转换为csv格式，并使用FTP相关的网络协议写出到远程FTP服务器。 写入FTP文件内容存放的是一张逻辑意义上的二维表，例如CSV格式的文本信息。 2 功能与限制 FtpWriter实现了从DataX协议转为FTP文件功能，FTP文件本身是无结构化数据存储，FtpWriter如下几个方面约定: 支持且仅支持写入文本类型(不支持BLOB如视频数据)的文件，且要求文本中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 写出时不支持文本压缩。 支持多线程写入，每个线程写入不同子文件。 我们不能做到： 单个文件不能支持并发写入。 3 功能说明 3.1 配置样例 { "setting": {}, "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": {}, "writer": { "name": "ftpwriter", "parameter": { "protocol": "sftp", "host": "***", "port": 22, "username": "xxx", "password": "xxx", "timeout": "60000", "connectPattern": "PASV", "path": "/tmp/data/", "fileName": "yixiao", "writeMode": "truncate|append|nonConflict", "fieldDelimiter": ",", "encoding": "UTF-8", "nullFormat": "null", "dateFormat": "yyyy-MM-dd", "fileFormat": "csv", "suffix": ".csv", "header": [] } } } ] } } 3.2 参数说明 protocol 描述：ftp服务器协议，目前支持传输协议有ftp和sftp。 必选：是 默认值：无 host 描述：ftp服务器地址。 必选：是 默认值：无 port 描述：ftp服务器端口。 必选：否 默认值：若传输协议是sftp协议，默认值是22；若传输协议是标准ftp协议，默认值是21 timeout 描述：连接ftp服务器连接超时时间，单位毫秒。 必选：否 默认值：60000（1分钟） username 描述：ftp服务器访问用户名。 必选：是 默认值：无 password 描述：ftp服务器访问密码。 必选：是 默认值：无 path 描述：FTP文件系统的路径信息，FtpWriter会写入Path目录下属多个文件。 必选：是 默认值：无 fileName 描述：FtpWriter写入的文件名，该文件名会添加随机的后缀作为每个线程写入实际文件名。 必选：是 默认值：无 writeMode 描述：FtpWriter写入前数据清理处理模式： truncate，写入前清理目录下一fileName前缀的所有文件。 append，写入前不做任何处理，DataX FtpWriter直接使用filename写入，并保证文件名不冲突。 nonConflict，如果目录下有fileName前缀的文件，直接报错。 必选：是 默认值：无 fieldDelimiter 描述：读取的字段分隔符 必选：否 默认值：, compress 描述：文本压缩类型，暂时不支持。 必选：否 默认值：无压缩 encoding 描述：读取文件的编码配置。 必选：否 默认值：utf-8 nullFormat 描述：文本文件中无法使用标准字符串定义null(空指针)，DataX提供nullFormat定义哪些字符串可以表示为null。 例如如果用户配置: nullFormat="\N"，那么如果源头数据是"\N"，DataX视作null字段。 必选：否 默认值：\N dateFormat 描述：日期类型的数据序列化到文件中时的格式，例如 “dateFormat”: “yyyy-MM-dd”。 必选：否 默认值：无 fileFormat 描述：文件写出的格式，包括csv (http://zh.wikipedia.org/wiki/%E9%80%97%E5%8F%B7%E5%88%86%E9%9A%94%E5%80%BC) 和text两种，csv是严格的csv格式，如果待写数据包括列分隔符，则会按照csv的转义语法转义，转义符号为双引号"；text格式是用列分隔符简单分割待写数据，对于待写数据包括列分隔符情况下不做转义。 必选：否 默认值：text suffix 描述：最后输出文件的后缀，当前支持 “.text"以及”.csv" 必选：否 默认值："" header 描述：txt写出时的表头，示例[‘id’, ’name’, ‘age’]。 必选：否 默认值：无 ...

DataX GDBReader 1. 快速介绍 GDBReader插件实现读取GDB实例数据的功能，通过Gremlin Client连接远程GDB实例，按配置提供的label生成查询DSL，遍历点或边数据，包括属性数据，并将数据写入到Record中给到Writer使用。 2. 实现原理 GDBReader使用Gremlin Client连接GDB实例，按label分不同Task取点或边数据。 单个Task中按label遍历点或边的id，再切分范围分多次请求查询点或边和属性数据，最后将点或边数据根据配置转换成指定格式记录发送给下游写插件。 GDBReader按label切分多个Task并发，同一个label的数据批量异步获取来加快读取速度。如果配置读取的label列表为空，任务启动前会从GDB查询所有label再切分Task。 3. 功能说明 GDB中点和边不同，读取需要区分点和边点配置。 3.1 点配置样例 { "job": { "setting": { "speed": { "channel": 1 } "errorLimit": { "record": 1 } }, "content": [ { "reader": { "name": "gdbreader", "parameter": { "host": "10.218.145.24", "port": 8182, "username": "***", "password": "***", "fetchBatchSize": 100, "rangeSplitSize": 1000, "labelType": "VERTEX", "labels": ["label1", "label2"], "column": [ { "name": "id", "type": "string", "columnType": "primaryKey" }, { "name": "label", "type": "string", "columnType": "primaryLabel" }, { "name": "age", "type": "int", "columnType": "vertexProperty" } ] } }, "writer": { "name": "streamwriter", "parameter": { "print": true } } } ] } } 3.2 边配置样例 { "job": { "setting": { "speed": { "channel": 1 }, "errorLimit": { "record": 1 } }, "content": [ { "reader": { "name": "gdbreader", "parameter": { "host": "10.218.145.24", "port": 8182, "username": "***", "password": "***", "fetchBatchSize": 100, "rangeSplitSize": 1000, "labelType": "EDGE", "labels": ["label1", "label2"], "column": [ { "name": "id", "type": "string", "columnType": "primaryKey" }, { "name": "label", "type": "string", "columnType": "primaryLabel" }, { "name": "srcId", "type": "string", "columnType": "srcPrimaryKey" }, { "name": "srcLabel", "type": "string", "columnType": "srcPrimaryLabel" }, { "name": "dstId", "type": "string", "columnType": "srcPrimaryKey" }, { "name": "dstLabel", "type": "string", "columnType": "srcPrimaryLabel" }, { "name": "name", "type": "string", "columnType": "edgeProperty" }, { "name": "weight", "type": "double", "columnType": "edgeProperty" } ] } }, "writer": { "name": "streamwriter", "parameter": { "print": true } } } ] } } 3.3 参数说明 host 描述：GDB实例连接地址，对应’实例管理’->‘基本信息’页面的网络地址 必选：是 默认值：无 port 描述：GDB实例连接地址对应的端口 必选：是 默认值：8182 username 描述：GDB实例账号名 必选：是 默认值：无 password 描述：GDB实例账号名对应的密码 必选：是 默认值：无 fetchBatchSize 描述：一次GDB请求读取点或边的数量，响应包含点或边以及属性 必选：是 默认值：100 rangeSplitSize 描述：id遍历，一次遍历请求扫描的id个数 必选：是 默认值：10 * fetchBatchSize labels 描述：标签数组，即需要导出的点或边标签，支持读取多个标签，用数组表示。如果留空([])，表示GDB中所有点或边标签 必选：是 默认值：无 labelType 描述：数据标签类型，支持点、边两种枚举值 VERTEX：表示点 EDGE：表示边 必选：是 默认值：无 column 描述：点或边字段映射关系配置 必选：是 默认值：无 column -> name 描述：点或边映射关系的字段名，指定属性时表示读取的属性名，读取其他字段时会被忽略 必选：是 默认值：无 column -> type 描述：点或边映射关系的字段类型 id, label在GDB中都是string类型，配置非string类型时可能会转换失败 普通属性支持基础类型，包括int, long, float, double, boolean, string GDBReader尽量将读取到的数据转换成配置要求的类型，但转换失败会导致该条记录错误 必选：是 默认值：无 column -> columnType 描述：GDB点或边数据到列数据的映射关系，支持以下枚举值： primaryKey： 表示该字段是点或边的id primaryLabel： 表示该字段是点或边的label srcPrimaryKey： 表示该字段是边关联的起点id，只在读取边时使用 srcPrimaryLabel： 表示该字段是边关联的起点label，只在读取边时使用 dstPrimaryKey： 表示该字段是边关联的终点id，只在读取边时使用 dstPrimaryLabel： 表示该字段是边关联的终点label，只在读取边时使用 vertexProperty： 表示该字段是点的属性，只在读取点时使用，应用到SET属性时只读取其中的一个属性值 vertexJsonProperty： 表示该字段是点的属性集合，只在读取点时使用。属性集合使用JSON格式输出，包含所有的属性，不能与其他vertexProperty配置一起使用 edgeProperty： 表示该字段是边的属性，只在读取边时使用 edgeJsonProperty： 表示该字段是边的属性集合，只在读取边时使用。属性集合使用JSON格式输出，包含所有的属性，不能与其他edgeProperty配置一起使用 必选：是 默认值：无 vertexJsonProperty格式示例，新增c字段区分SET属性，但是SET属性只包含单个属性值时会标记成普通属性 {"properties":[ {"k":"name","t","string","v":"Jack","c":"set"}, {"k":"name","t","string","v":"Luck","c":"set"}, {"k":"age","t","int","v":"20","c":"single"} ]} edgeJsonProperty格式示例，边不支持多值属性 {"properties":[ {"k":"created_at","t","long","v":"153498653"}, {"k":"weight","t","double","v":"3.14"} ]} ...

DataX GDBWriter 1 快速介绍 GDBWriter插件实现了写入数据到GDB实例的功能。GDBWriter通过Gremlin Client连接远程GDB实例，获取Reader的数据，生成写入DSL语句，将数据写入到GDB。 2 实现原理 GDBWriter通过DataX框架获取Reader生成的协议数据，使用g.addV/E(GDB___label).property(id, GDB___id).property(GDB___PK1, GDB___PV1)...语句写入数据到GDB实例。 可以配置Gremlin Client工作在session模式，由客户端控制事务，在一次事务中实现多个记录的批量写入。 3 功能说明 因为GDB中点和边的配置不同，导入时需要区分点和边的配置。 3.1 点配置样例 这里是一份从内存生成点数据导入GDB实例的配置 { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column" : [ { "random": "1,100", "type": "double" }, { "random": "1000,1200", "type": "long" }, { "random": "60,64", "type": "string" }, { "random": "100,1000", "type": "long" }, { "random": "32,48", "type": "string" } ], "sliceRecordCount": 1000 } }, "writer": { "name": "gdbwriter", "parameter": { "host": "gdb-endpoint", "port": 8182, "username": "root", "password": "***", "writeMode": "INSERT", "labelType": "VERTEX", "label": "#{1}", "idTransRule": "none", "session": true, "maxRecordsInBatch": 64, "column": [ { "name": "id", "value": "#{0}", "type": "string", "columnType": "primaryKey" }, { "name": "vertex_propKey", "value": "#{2}", "type": "string", "columnType": "vertexSetProperty" }, { "name": "vertex_propKey", "value": "#{3}", "type": "long", "columnType": "vertexSetProperty" }, { "name": "vertex_propKey2", "value": "#{4}", "type": "string", "columnType": "vertexProperty" } ] } } } ] } } 3.2 边配置样例 这里是一份从内存生成边数据导入GDB实例的配置 注意 下面配置导入边时，需要提前在GDB实例中写入点，要求分别存在id为person-{{i}}和book-{{i}}的点，其中i取值0~100。 ...

DataX HdfsReader 插件文档 1 快速介绍 HdfsReader提供了读取分布式文件系统数据存储的能力。在底层实现上，HdfsReader获取分布式文件系统上文件的数据，并转换为DataX传输协议传递给Writer。 目前HdfsReader支持的文件格式有textfile（text）、orcfile（orc）、rcfile（rc）、sequence file（seq）和普通逻辑二维表（csv）类型格式的文件，且文件内容存放的必须是一张逻辑意义上的二维表。 HdfsReader需要Jdk1.7及以上版本的支持。 2 功能与限制 HdfsReader实现了从Hadoop分布式文件系统Hdfs中读取文件数据并转为DataX协议的功能。textfile是Hive建表时默认使用的存储格式，数据不做压缩，本质上textfile就是以文本的形式将数据存放在hdfs中，对于DataX而言，HdfsReader实现上类比TxtFileReader，有诸多相似之处。orcfile，它的全名是Optimized Row Columnar file，是对RCFile做了优化。据官方文档介绍，这种文件格式可以提供一种高效的方法来存储Hive数据。HdfsReader利用Hive提供的OrcSerde类，读取解析orcfile文件的数据。目前HdfsReader支持的功能如下： 支持textfile、orcfile、rcfile、sequence file和csv格式的文件，且要求文件内容存放的是一张逻辑意义上的二维表。 支持多种类型数据读取(使用String表示)，支持列裁剪，支持列常量 支持递归读取、支持正则表达式（"*“和”?"）。 支持orcfile数据压缩，目前支持SNAPPY，ZLIB两种压缩方式。 多个File可以支持并发读取。 支持sequence file数据压缩，目前支持lzo压缩方式。 csv类型支持压缩格式有：gzip、bz2、zip、lzo、lzo_deflate、snappy。 目前插件中Hive版本为1.1.1，Hadoop版本为2.7.1（Apache［为适配JDK1.7］,在Hadoop 2.5.0, Hadoop 2.6.0 和Hive 1.2.0测试环境中写入正常；其它版本需后期进一步测试； 支持kerberos认证（注意：如果用户需要进行kerberos认证，那么用户使用的Hadoop集群版本需要和hdfsreader的Hadoop版本保持一致，如果高于hdfsreader的Hadoop版本，不保证kerberos认证有效） 我们暂时不能做到： 单个File支持多线程并发读取，这里涉及到单个File内部切分算法。二期考虑支持。 目前还不支持hdfs HA; 3 功能说明 3.1 配置样例 { "job": { "setting": { "speed": { "channel": 3 } }, "content": [ { "reader": { "name": "hdfsreader", "parameter": { "path": "/user/hive/warehouse/mytable01/*", "defaultFS": "hdfs://xxx:port", "column": [ { "index": 0, "type": "long" }, { "index": 1, "type": "boolean" }, { "type": "string", "value": "hello" }, { "index": 2, "type": "double" } ], "fileType": "orc", "encoding": "UTF-8", "fieldDelimiter": "," } }, "writer": { "name": "streamwriter", "parameter": { "print": true } } } ] } } 3.2 参数说明（各个配置项值前后不允许有空格） path 描述：要读取的文件路径，如果要读取多个文件，可以使用正则表达式"*"，注意这里可以支持填写多个路径。。 当指定单个Hdfs文件，HdfsReader暂时只能使用单线程进行数据抽取。二期考虑在非压缩文件情况下针对单个File可以进行多线程并发读取。 当指定多个Hdfs文件，HdfsReader支持使用多线程进行数据抽取。线程并发数通过通道数指定。 当指定通配符，HdfsReader尝试遍历出多个文件信息。例如: 指定/代表读取/目录下所有的文件，指定/bazhen/*代表读取bazhen目录下游所有的文件。HdfsReader目前只支持"“和”?“作为文件通配符。 特别需要注意的是，DataX会将一个作业下同步的所有的文件视作同一张数据表。用户必须自己保证所有的File能够适配同一套schema信息。并且提供给DataX权限可读。 必选：是 默认值：无 defaultFS 描述：Hadoop hdfs文件系统namenode节点地址。 目前HdfsReader已经支持Kerberos认证，如果需要权限认证，则需要用户配置kerberos参数，见下面 必选：是 默认值：无 fileType 描述：文件的类型，目前只支持用户配置为"text”、“orc”、“rc”、“seq”、“csv”。 text表示textfile文件格式 orc表示orcfile文件格式 rc表示rcfile文件格式 seq表示sequence file文件格式 csv表示普通hdfs文件格式（逻辑二维表） 特别需要注意的是，HdfsReader能够自动识别文件是orcfile、textfile或者还是其它类型的文件，但该项是必填项，HdfsReader则会只读取用户配置的类型的文件，忽略路径下其他格式的文件 另外需要注意的是，由于textfile和orcfile是两种完全不同的文件格式，所以HdfsReader对这两种文件的解析方式也存在差异，这种差异导致hive支持的复杂复合类型(比如map,array,struct,union)在转换为DataX支持的String类型时，转换的结果格式略有差异，比如以map类型为例： orcfile map类型经hdfsreader解析转换成datax支持的string类型后，结果为"{job=80, team=60, person=70}" textfile map类型经hdfsreader解析转换成datax支持的string类型后，结果为"job:80,team:60,person:70" 从上面的转换结果可以看出，数据本身没有变化，但是表示的格式略有差异，所以如果用户配置的文件路径中要同步的字段在Hive中是复合类型的话，建议配置统一的文件格式。 如果需要统一复合类型解析出来的格式，我们建议用户在hive客户端将textfile格式的表导成orcfile格式的表 必选：是 默认值：无 column 描述：读取字段列表，type指定源数据的类型，index指定当前列来自于文本第几列(以0开始)，value指定当前类型为常量，不从源头文件读取数据，而是根据value值自动生成对应的列。 默认情况下，用户可以全部按照String类型读取数据，配置如下： "column": ["*"] 用户可以指定Column字段信息，配置如下： ...

DataX HdfsWriter 插件文档 1 快速介绍 HdfsWriter提供向HDFS文件系统指定路径中写入TEXTFile文件和ORCFile文件,文件内容可与hive中表关联。 2 功能与限制 (1)、目前HdfsWriter仅支持textfile和orcfile两种格式的文件，且文件内容存放的必须是一张逻辑意义上的二维表; (2)、由于HDFS是文件系统，不存在schema的概念，因此不支持对部分列写入; (3)、目前仅支持与以下Hive数据类型： 数值型：TINYINT,SMALLINT,INT,BIGINT,FLOAT,DOUBLE 字符串类型：STRING,VARCHAR,CHAR 布尔类型：BOOLEAN 时间类型：DATE,TIMESTAMP 目前不支持：decimal、binary、arrays、maps、structs、union类型; (4)、对于Hive分区表目前仅支持一次写入单个分区; (5)、对于textfile需用户保证写入hdfs文件的分隔符与在Hive上创建表时的分隔符一致,从而实现写入hdfs数据与Hive表字段关联; (6)、HdfsWriter实现过程是：首先根据用户指定的path，创建一个hdfs文件系统上不存在的临时目录，创建规则：path_随机；然后将读取的文件写入这个临时目录；全部写入后再将这个临时目录下的文件移动到用户指定目录（在创建文件时保证文件名不重复）; 最后删除临时目录。如果在中间过程发生网络中断等情况造成无法与hdfs建立连接，需要用户手动删除已经写入的文件和临时目录。 (7)、目前插件中Hive版本为1.1.1，Hadoop版本为2.7.1（Apache［为适配JDK1.7］,在Hadoop 2.5.0, Hadoop 2.6.0 和Hive 1.2.0测试环境中写入正常；其它版本需后期进一步测试； (8)、目前HdfsWriter支持Kerberos认证（注意：如果用户需要进行kerberos认证，那么用户使用的Hadoop集群版本需要和hdfsreader的Hadoop版本保持一致，如果高于hdfsreader的Hadoop版本，不保证kerberos认证有效） 3 功能说明 3.1 配置样例 { "setting": {}, "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "txtfilereader", "parameter": { "path": ["/Users/shf/workplace/txtWorkplace/job/dataorcfull.txt"], "encoding": "UTF-8", "column": [ { "index": 0, "type": "long" }, { "index": 1, "type": "long" }, { "index": 2, "type": "long" }, { "index": 3, "type": "long" }, { "index": 4, "type": "DOUBLE" }, { "index": 5, "type": "DOUBLE" }, { "index": 6, "type": "STRING" }, { "index": 7, "type": "STRING" }, { "index": 8, "type": "STRING" }, { "index": 9, "type": "BOOLEAN" }, { "index": 10, "type": "date" }, { "index": 11, "type": "date" } ], "fieldDelimiter": "\t" } }, "writer": { "name": "hdfswriter", "parameter": { "defaultFS": "hdfs://xxx:port", "fileType": "orc", "path": "/user/hive/warehouse/writerorc.db/orcfull", "fileName": "xxxx", "column": [ { "name": "col1", "type": "TINYINT" }, { "name": "col2", "type": "SMALLINT" }, { "name": "col3", "type": "INT" }, { "name": "col4", "type": "BIGINT" }, { "name": "col5", "type": "FLOAT" }, { "name": "col6", "type": "DOUBLE" }, { "name": "col7", "type": "STRING" }, { "name": "col8", "type": "VARCHAR" }, { "name": "col9", "type": "CHAR" }, { "name": "col10", "type": "BOOLEAN" }, { "name": "col11", "type": "date" }, { "name": "col12", "type": "TIMESTAMP" } ], "writeMode": "append", "fieldDelimiter": "\t", "compress":"NONE" } } } ] } } 3.2 参数说明 defaultFS 描述：Hadoop hdfs文件系统namenode节点地址。格式：hdfs://ip:端口；例如：hdfs://127.0.0.1:9000 必选：是 默认值：无 fileType 描述：文件的类型，目前只支持用户配置为"text"或"orc"。 text表示textfile文件格式 orc表示orcfile文件格式 必选：是 默认值：无 path 描述：存储到Hadoop hdfs文件系统的路径信息，HdfsWriter会根据并发配置在Path目录下写入多个文件。为与hive表关联，请填写hive表在hdfs上的存储路径。例：Hive上设置的数据仓库的存储路径为：/user/hive/warehouse/ ，已建立数据库：test，表：hello；则对应的存储路径为：/user/hive/warehouse/test.db/hello 必选：是 默认值：无 fileName 描述：HdfsWriter写入时的文件名，实际执行时会在该文件名后添加随机的后缀作为每个线程写入实际文件名。 必选：是 默认值：无 column 描述：写入数据的字段，不支持对部分列写入。为与hive中表关联，需要指定表中所有字段名和字段类型，其中：name指定字段名，type指定字段类型。 用户可以指定Column字段信息，配置如下： "column": [ { "name": "userName", "type": "string" }, { "name": "age", "type": "long" } ] 必选：是 默认值：无 writeMode 描述：hdfswriter写入前数据清理处理模式： append，写入前不做任何处理，DataX hdfswriter直接使用filename写入，并保证文件名不冲突。 nonConflict，如果目录下有fileName前缀的文件，直接报错。 必选：是 默认值：无 fieldDelimiter 描述：hdfswriter写入时的字段分隔符,需要用户保证与创建的Hive表的字段分隔符一致，否则无法在Hive表中查到数据 必选：是 默认值：无 compress 描述：hdfs文件压缩类型，默认不填写意味着没有压缩。其中：text类型文件支持压缩类型有gzip、bzip2;orc类型文件支持的压缩类型有NONE、SNAPPY（需要用户安装SnappyCodec）。 必选：否 默认值：无压缩 hadoopConfig 描述：hadoopConfig里可以配置与Hadoop相关的一些高级参数，比如HA的配置。 "hadoopConfig":{ "dfs.nameservices": "testDfs", "dfs.ha.namenodes.testDfs": "namenode1,namenode2", "dfs.namenode.rpc-address.aliDfs.namenode1": "", "dfs.namenode.rpc-address.aliDfs.namenode2": "", "dfs.client.failover.proxy.provider.testDfs": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider" } 必选：否 默认值：无 encoding 描述：写文件的编码配置。 必选：否 默认值：utf-8，慎重修改 haveKerberos 描述：是否有Kerberos认证，默认false 例如如果用户配置true，则配置项kerberosKeytabFilePath，kerberosPrincipal为必填。 必选：haveKerberos 为true必选 默认值：false kerberosKeytabFilePath 描述：Kerberos认证 keytab文件路径，绝对路径 必选：否 默认值：无 kerberosPrincipal 描述：Kerberos认证Principal名，如xxxx/hadoopclient@xxx.xxx 必选：haveKerberos 为true必选 默认值：无 ...

DataX KingbaseesWriter 1 快速介绍 KingbaseesWriter插件实现了写入数据到 KingbaseES主库目的表的功能。在底层实现上，KingbaseesWriter通过JDBC连接远程 KingbaseES 数据库，并执行相应的 insert into … sql 语句将数据写入 KingbaseES，内部会分批次提交入库。 KingbaseesWriter面向ETL开发工程师，他们使用KingbaseesWriter从数仓导入数据到KingbaseES。同时 KingbaseesWriter亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 KingbaseesWriter通过 DataX 框架获取 Reader 生成的协议数据，根据你配置生成相应的SQL插入语句 insert into...(当主键/唯一性索引冲突时会写不进去冲突的行) 注意： 1. 目的表所在数据库必须是主库才能写入数据；整个任务至少需具备 insert into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 2. KingbaseesWriter和MysqlWriter不同，不支持配置writeMode参数。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 KingbaseesWriter导入的数据。 { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column" : [ { "value": "DataX", "type": "string" }, { "value": 19880808, "type": "long" }, { "value": "1988-08-08 08:08:08", "type": "date" }, { "value": true, "type": "bool" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 1000 } }, "writer": { "name": "kingbaseeswriter", "parameter": { "username": "xx", "password": "xx", "column": [ "id", "name" ], "preSql": [ "delete from test" ], "connection": [ { "jdbcUrl": "jdbc:kingbase8://127.0.0.1:3002/datax", "table": [ "test" ] } ] } } } ] } } 3.2 参数说明 jdbcUrl 描述：目的数据库的 JDBC 连接信息 ,jdbcUrl必须包含在connection配置单元中。 注意：1、在一个数据库上只能配置一个值。 2、jdbcUrl按照KingbaseES官方规范，并可以填写连接附加参数信息。具体请参看KingbaseES官方文档或者咨询对应 DBA。 必选：是 默认值：无 username 描述：目的数据库的用户名 必选：是 默认值：无 password 描述：目的数据库的密码 必选：是 默认值：无 table 描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。 注意：table 和 jdbcUrl 必须包含在 connection 配置单元中 必选：是 默认值：无 column 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: “column”: [“id”,“name”,“age”]。如果要依次写入全部列，使用*表示, 例如: “column”: ["*"] 注意：1、我们强烈不推荐你这样配置，因为当你目的表字段个数、类型等有改动时，你的任务可能运行不正确或者失败 2、此处 column 不能配置任何常量值 必选：是 默认值：否 preSql 描述：写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 @table 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, … datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置："preSql":["delete from @table"]，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称 必选：否 默认值：无 postSql 描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql ） 必选：否 默认值：无 batchSize 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与KingbaseES的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。 必选：否 默认值：1024 ...

Datax MongoDBReader 1 快速介绍 MongoDBReader 插件利用 MongoDB 的java客户端MongoClient进行MongoDB的读操作。最新版本的Mongo已经将DB锁的粒度从DB级别降低到document级别，配合上MongoDB强大的索引功能，基本可以达到高性能的读取MongoDB的需求。 2 实现原理 MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的JOB程序按照指定的规则对MongoDB中的数据进行分片，并行读取，然后将MongoDB支持的类型通过逐一判断转换成Datax支持的类型。 3 功能说明 该示例从ODPS读一份数据到MongoDB。 { "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "mongodbreader", "parameter": { "address": ["127.0.0.1:27017"], "userName": "", "userPassword": "", "dbName": "tag_per_data", "collectionName": "tag_data12", "column": [ { "name": "unique_id", "type": "string" }, { "name": "sid", "type": "string" }, { "name": "user_id", "type": "string" }, { "name": "auction_id", "type": "string" }, { "name": "content_type", "type": "string" }, { "name": "pool_type", "type": "string" }, { "name": "frontcat_id", "type": "Array", "spliter": "" }, { "name": "categoryid", "type": "Array", "spliter": "" }, { "name": "gmt_create", "type": "string" }, { "name": "taglist", "type": "Array", "spliter": " " }, { "name": "property", "type": "string" }, { "name": "scorea", "type": "int" }, { "name": "scoreb", "type": "int" }, { "name": "scorec", "type": "int" } ] } }, "writer": { "name": "odpswriter", "parameter": { "project": "tb_ai_recommendation", "table": "jianying_tag_datax_read_test01", "column": [ "unique_id", "sid", "user_id", "auction_id", "content_type", "pool_type", "frontcat_id", "categoryid", "gmt_create", "taglist", "property", "scorea", "scoreb" ], "accessId": "**************", "accessKey": "********************", "truncate": true, "odpsServer": "xxx/api", "tunnelServer": "xxx", "accountType": "aliyun" } } } ] } } 4 参数说明 address： MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】 userName：MongoDB的用户名。【选填】 userPassword： MongoDB的密码。【选填】 collectionName： MonogoDB的集合名。【必填】 column：MongoDB的文档列名。【必填】 name：Column的名字。【必填】 type：Column的类型。【选填】 splitter：因为MongoDB支持数组类型，但是Datax框架本身不支持数组类型，所以mongoDB读出来的数组类型要通过这个分隔符合并成字符串。【选填】 query: MongoDB的额外查询条件。【选填】 5 类型转换 DataX 内部类型 MongoDB 数据类型 Long int, Long Double double String string, array Date date Boolean boolean Bytes bytes 6 性能报告 7 测试报告

Datax MongoDBWriter 1 快速介绍 MongoDBWriter 插件利用 MongoDB 的java客户端MongoClient进行MongoDB的写操作。最新版本的Mongo已经将DB锁的粒度从DB级别降低到document级别，配合上MongoDB强大的索引功能，基本可以满足数据源向MongoDB写入数据的需求，针对数据更新的需求，通过配置业务主键的方式也可以实现。 2 实现原理 MongoDBWriter通过Datax框架获取Reader生成的数据，然后将Datax支持的类型通过逐一判断转换成MongoDB支持的类型。其中一个值得指出的点就是Datax本身不支持数组类型，但是MongoDB支持数组类型，并且数组类型的索引还是蛮强大的。为了使用MongoDB的数组类型，则可以通过参数的特殊配置，将字符串可以转换成MongoDB中的数组。类型转换之后，就可以依托于Datax框架并行的写入MongoDB。 3 功能说明 该示例从ODPS读一份数据到MongoDB。 { "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "odpsreader", "parameter": { "accessId": "********", "accessKey": "*********", "project": "tb_ai_recommendation", "table": "jianying_tag_datax_test", "column": [ "unique_id", "sid", "user_id", "auction_id", "content_type", "pool_type", "frontcat_id", "categoryid", "gmt_create", "taglist", "property", "scorea", "scoreb" ], "splitMode": "record", "odpsServer": "http://xxx/api" } }, "writer": { "name": "mongodbwriter", "parameter": { "address": [ "127.0.0.1:27017" ], "userName": "", "userPassword": "", "dbName": "tag_per_data", "collectionName": "tag_data", "column": [ { "name": "unique_id", "type": "string" }, { "name": "sid", "type": "string" }, { "name": "user_id", "type": "string" }, { "name": "auction_id", "type": "string" }, { "name": "content_type", "type": "string" }, { "name": "pool_type", "type": "string" }, { "name": "frontcat_id", "type": "Array", "splitter": " " }, { "name": "categoryid", "type": "Array", "splitter": " " }, { "name": "gmt_create", "type": "string" }, { "name": "taglist", "type": "Array", "splitter": " " }, { "name": "property", "type": "string" }, { "name": "scorea", "type": "int" }, { "name": "scoreb", "type": "int" }, { "name": "scorec", "type": "int" } ], "upsertInfo": { "isUpsert": "true", "upsertKey": "unique_id" } } } } ] } } 4 参数说明 address： MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】 userName：MongoDB的用户名。【选填】 userPassword： MongoDB的密码。【选填】 collectionName： MonogoDB的集合名。【必填】 column：MongoDB的文档列名。【必填】 name：Column的名字。【必填】 type：Column的类型。【选填】 splitter：特殊分隔符，当且仅当要处理的字符串要用分隔符分隔为字符数组时，才使用这个参数，通过这个参数指定的分隔符，将字符串分隔存储到MongoDB的数组中。【选填】 upsertInfo：指定了传输数据时更新的信息。【选填】 isUpsert：当设置为true时，表示针对相同的upsertKey做更新操作。【选填】 upsertKey：upsertKey指定了没行记录的业务主键。用来做更新时使用。【选填】 5 类型转换 DataX 内部类型 MongoDB 数据类型 Long int, Long Double double String string, array Date date Boolean boolean Bytes bytes 6 性能报告 7 测试报告

DataX MysqlWriter 1 快速介绍 MysqlWriter 插件实现了写入数据到 Mysql 主库的目的表的功能。在底层实现上， MysqlWriter 通过 JDBC 连接远程 Mysql 数据库，并执行相应的 insert into … 或者 ( replace into …) 的 sql 语句将数据写入 Mysql，内部会分批次提交入库，需要数据库本身采用 innodb 引擎。 MysqlWriter 面向ETL开发工程师，他们使用 MysqlWriter 从数仓导入数据到 Mysql。同时 MysqlWriter 亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 MysqlWriter 通过 DataX 框架获取 Reader 生成的协议数据，根据你配置的 writeMode 生成 insert into...(当主键/唯一性索引冲突时会写不进去冲突的行) 或者 replace into...(没有遇到主键/唯一性索引冲突时，与 insert into 行为一致，冲突时会用新行替换原有行所有字段) 的语句写入数据到 Mysql。出于性能考虑，采用了 PreparedStatement + Batch，并且设置了：rewriteBatchedStatements=true，将数据缓冲到线程上下文 Buffer 中，当 Buffer 累计到预定阈值时，才发起写入请求。 注意：目的表所在数据库必须是主库才能写入数据；整个任务至少需要具备 insert/replace into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 Mysql 导入的数据。 { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column" : [ { "value": "DataX", "type": "string" }, { "value": 19880808, "type": "long" }, { "value": "1988-08-08 08:08:08", "type": "date" }, { "value": true, "type": "bool" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 1000 } }, "writer": { "name": "mysqlwriter", "parameter": { "writeMode": "insert", "username": "root", "password": "root", "column": [ "id", "name" ], "session": [ "set session sql_mode='ANSI'" ], "preSql": [ "delete from test" ], "connection": [ { "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/datax?useUnicode=true&characterEncoding=gbk", "table": [ "test" ] } ] } } } ] } } 3.2 参数说明 jdbcUrl 描述：目的数据库的 JDBC 连接信息。作业运行时，DataX 会在你提供的 jdbcUrl 后面追加如下属性：yearIsDateType=false&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true 注意：1、在一个数据库上只能配置一个 jdbcUrl 值。这与 MysqlReader 支持多个备库探测不同，因为此处不支持同一个数据库存在多个主库的情况(双主导入数据情况) 2、jdbcUrl按照Mysql官方规范，并可以填写连接附加控制信息，比如想指定连接编码为 gbk ，则在 jdbcUrl 后面追加属性 useUnicode=true&characterEncoding=gbk。具体请参看 Mysql官方文档或者咨询对应 DBA。 必选：是 默认值：无 username 描述：目的数据库的用户名 必选：是 默认值：无 password 描述：目的数据库的密码 必选：是 默认值：无 table 描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。 注意：table 和 jdbcUrl 必须包含在 connection 配置单元中 必选：是 默认值：无 column 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: “column”: [“id”,“name”,“age”]。如果要依次写入全部列，使用*表示, 例如: "column": ["*"]。 **column配置项必须指定，不能留空！** 注意：1、我们强烈不推荐你这样配置，因为当你目的表字段个数、类型等有改动时，你的任务可能运行不正确或者失败 2、 column 不能配置任何常量值 必选：是 默认值：否 session 描述: DataX在获取Mysql连接时，执行session指定的SQL语句，修改当前connection session属性 必须: 否 默认值: 空 preSql 描述：写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 @table 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, … datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置："preSql":["delete from 表名"]，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称 必选：否 默认值：无 postSql 描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql ） 必选：否 默认值：无 writeMode 描述：控制写入数据到目标表采用 insert into 或者 replace into 或者 ON DUPLICATE KEY UPDATE 语句 必选：是 所有选项：insert/replace/update 默认值：insert batchSize 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与Mysql的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。 必选：否 默认值：1024 ...

DataX OCSWriter 适用memcached客户端写入ocs 1 快速介绍 1.1 OCS简介 开放缓存服务( Open Cache Service，简称OCS）是基于内存的缓存服务，支持海量小数据的高速访问。OCS可以极大缓解对后端存储的压力，提高网站或应用的响应速度。OCS支持Key-Value的数据结构，兼容Memcached协议的客户端都可与OCS通信。 OCS 支持即开即用的方式快速部署；对于动态Web、APP应用，可通过缓存服务减轻对数据库的压力，从而提高网站整体的响应速度。 与本地MemCache相同之处在于OCS兼容Memcached协议，与用户环境兼容，可直接用于OCS服务 不同之处在于硬件和数据部署在云端，有完善的基础设施、网络安全保障、系统维护服务。所有的这些服务，都不需要投资，只需根据使用量进行付费即可。 1.2 OCSWriter简介 OCSWriter是DataX实现的，基于Memcached协议的数据写入OCS通道。 2 功能说明 2.1 配置样例 这里使用一份从内存产生的数据导入到OCS。 { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ { "value": "DataX", "type": "string" }, { "value": 19880808, "type": "long" }, { "value": "1988-08-08 08:08:08", "type": "date" }, { "value": true, "type": "bool" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 1000 } }, "writer": { "name": "ocswriter", "parameter": { "proxy": "xxxx", "port": "11211", "userName": "user", "password": "******", "writeMode": "set|add|replace|append|prepend", "writeFormat": "text|binary", "fieldDelimiter": "\u0001", "expireTime": 1000, "indexes": "0,2", "batchSize": 1000 } } } ] } } 2.2 参数说明 proxy 描述：OCS机器的ip或host。 必选：是 port 描述：OCS的连接域名，默认为11211 必选：否 默认值：11211 username 描述：OCS连接的访问账号。 必选：是 password 描述：OCS连接的访问密码 必选：是 writeMode 描述: OCSWriter写入方式，具体为： set: 存储这个数据，如果已经存在则覆盖 add: 存储这个数据，当且仅当这个key不存在的时候 replace: 存储这个数据，当且仅当这个key存在 append: 将数据存放在已存在的key对应的内容的后面，忽略exptime prepend: 将数据存放在已存在的key对应的内容的前面，忽略exptime 必选：是 writeFormat 描述: OCSWriter写出数据格式，目前支持两类数据写入方式: text: 将源端数据序列化为文本格式，其中第一个字段作为OCS写入的KEY，后续所有字段序列化为STRING类型，使用用户指定的fieldDelimiter作为间隔符，将文本拼接为完整的字符串再写入OCS。 binary: 将源端数据作为二进制直接写入，这类场景为未来做扩展使用，目前不支持。如果填写binary将会报错！ 必选：否 默认值：text expireTime 描述: OCS值缓存失效时间，目前MemCache支持两类过期时间， Unix时间(自1970.1.1开始到现在的秒数)，该时间指定了到未来某个时刻数据失效。 相对当前时间的秒数，该时间指定了从现在开始多长时间后数据失效。 注意：如果过期时间的秒数大于606024*30(即30天)，则服务端认为是Unix时间。 单位：秒 必选：否 默认值：0【0表示永久有效】 indexes 描述: 用数据的第几列当做ocs的key 必选：否 默认值：0 batchSize 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与OCS的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况[memcached版本暂不支持批量写]。 必选：否 默认值：256 fieldDelimiter 描述：写入ocs的key和value分隔符。比如：key=tom\u0001boston, value=28\u0001lawer\u0001male\u0001married 必选：否 默认值：\u0001 ...