DataX ODPSReader 1 快速介绍 ODPSReader 实现了从 ODPS读取数据的功能，有关ODPS请参看(https://help.aliyun.com/document_detail/27800.html?spm=5176.doc27803.6.101.NxCIgY)。 在底层实现上，ODPSReader 根据你配置的 源头项目 / 表 / 分区 / 表字段 等信息，通过 Tunnel 从 ODPS 系统中读取数据。 注意 1、如果你需要使用ODPSReader/Writer插件，由于 AccessId/AccessKey 解密的需要，请务必使用 JDK 1.6.32 及以上版本。JDK 安装事项，请联系 PE 处理 2、ODPSReader 不是通过 ODPS SQL （select ... from ... where ... ）来抽取数据的 3、注意区分你要读取的表是线上环境还是线下环境 4、目前 DataX3 依赖的 SDK 版本是： <dependency> <groupId>com.aliyun.odps</groupId> <artifactId>odps-sdk-core-internal</artifactId> <version>0.13.2</version> </dependency> 2 实现原理 ODPSReader 支持读取分区表、非分区表，不支持读取虚拟视图。当要读取分区表时，需要指定出具体的分区配置，比如读取 t0 表，其分区为 pt=1,ds=hangzhou 那么你需要在配置中配置该值。当要读取非分区表时，你不能提供分区配置。表字段可以依序指定全部列，也可以指定部分列，或者调整列顺序，或者指定常量字段，但是表字段中不能指定分区列（分区列不是表字段）。 注意：要特别注意 odpsServer、project、table、accessId、accessKey 的配置，因为直接影响到是否能够加载到你需要读取数据的表。很多权限问题都出现在这里。 3 功能说明 3.1 配置样例 这里使用一份读出 ODPS 数据然后打印到屏幕的配置样板。 { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { "name": "odpsreader", "parameter": { "accessId": "accessId", "accessKey": "accessKey", "project": "targetProjectName", "table": "tableName", "partition": [ "pt=1,ds=hangzhou" ], "column": [ "customer_id", "nickname" ], "packageAuthorizedProject": "yourCurrentProjectName", "splitMode": "record", "odpsServer": "http://xxx/api", "tunnelServer": "http://dt.odps.aliyun.com" } }, "writer": { "name": "streamwriter", "parameter": { "fieldDelimiter": "\t", "print": "true" } } } ] } } 3.2 参数说明 参数 accessId 描述：ODPS系统登录ID 必选：是 默认值：无 accessKey 描述：ODPS系统登录Key 必选：是 默认值：无 project 描述：读取数据表所在的 ODPS 项目名称（大小写不敏感） 必选：是 默认值：无 table 描述：读取数据表的表名称（大小写不敏感） 必选：是 默认值：无 partition 描述：读取数据所在的分区信息，支持linux shell通配符，包括 * 表示0个或多个字符，?代表任意一个字符。例如现在有分区表 test，其存在 pt=1,ds=hangzhou pt=1,ds=shanghai pt=2,ds=hangzhou pt=2,ds=beijing 四个分区，如果你想读取 pt=1,ds=shanghai 这个分区的数据，那么你应该配置为: "partition":["pt=1,ds=shanghai"]； 如果你想读取 pt=1下的所有分区，那么你应该配置为: "partition":["pt=1,ds=* "]；如果你想读取整个 test 表的所有分区的数据，那么你应该配置为: "partition":["pt=*,ds=*"] 必选：如果表为分区表，则必填。如果表为非分区表，则不能填写 默认值：无 column 描述：读取 odps 源头表的列信息。例如现在有表 test，其字段为：id,name,age 如果你想依次读取 id,name,age 那么你应该配置为: "column":["id","name","age"] 或者配置为:"column"=["*"] 这里 * 表示依次读取表的每个字段，但是我们不推荐你配置抽取字段为 * ，因为当你的表字段顺序调整、类型变更或者个数增减，你的任务就会存在源头表列和目的表列不能对齐的风险，会直接导致你的任务运行结果不正确甚至运行失败。如果你想依次读取 name,id 那么你应该配置为: "coulumn":["name","id"] 如果你想在源头抽取的字段中添加常量字段(以适配目标表的字段顺序)，比如你想抽取的每一行数据值为 age 列对应的值,name列对应的值,常量日期值1988-08-08 08:08:08,id 列对应的值 那么你应该配置为:"column":["age","name","'1988-08-08 08:08:08'","id"] 即常量列首尾用符号' 包住即可，我们内部实现上识别常量是通过检查你配置的每一个字段，如果发现有字段首尾都有'，则认为其是常量字段，其实际值为去除' 之后的值。 注意：ODPSReader 抽取数据表不是通过 ODPS 的 Select SQL 语句，所以不能在字段上指定函数，也不能指定分区字段名称（分区字段不属于表字段） 必选：是 默认值：无 odpsServer 描述：源头表 所在 ODPS 系统的server 地址 必选：是 默认值：无 tunnelServer 描述：源头表 所在 ODPS 系统的tunnel 地址 必选：是 默认值：无 splitMode 描述：读取源头表时切分所需要的模式。默认值为 record，可不填，表示根据切分份数，按照记录数进行切分。如果你的任务目的端为 Mysql，并且是 Mysql 的多个表，那么根据现在 DataX 结构，你的源头表必须是分区表，并且每个分区依次对应目的端 Mysql 的多个分表，则此时应该配置为"splitMode":"partition" 必选：否 默认值：record accountProvider [待定] 描述：读取时使用的 ODPS 账号类型。目前支持 aliyun/taobao 两种类型。默认为 aliyun，可不填 必选：否 默认值：aliyun packageAuthorizedProject * 描述：被package授权的project，即用户当前所在project <br /> 必选：否 默认值：无 isCompress * 描述：是否压缩读取，bool类型: "true"表示压缩, "false"标示不压缩 <br /> 必选：否 默认值：“false” : 不压缩 ...

DataX ODPS写入 1 快速介绍 ODPSWriter插件用于实现往ODPS插入或者更新数据，主要提供给etl开发同学将业务数据导入odps，适合于TB,GB数量级的数据传输，如果需要传输PB量级的数据，请选择dt task工具 ; 2 实现原理 在底层实现上，ODPSWriter是通过DT Tunnel写入ODPS系统的，有关ODPS的更多技术细节请参看 ODPS主站 https://data.aliyun.com/product/odps 和ODPS产品文档 https://help.aliyun.com/product/27797.html 目前 DataX3 依赖的 SDK 版本是： <dependency> <groupId>com.aliyun.odps</groupId> <artifactId>odps-sdk-core-internal</artifactId> <version>0.13.2</version> </dependency> 注意: 如果你需要使用ODPSReader/Writer插件，请务必使用JDK 1.6-32及以上版本 使用java -version查看Java版本号 3 功能说明 3.1 配置样例 这里使用一份从内存产生到ODPS导入的数据。 { "job": { "setting": { "speed": { "byte": 1048576 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ { "value": "DataX", "type": "string" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 100000 } }, "writer": { "name": "odpswriter", "parameter": { "project": "chinan_test", "table": "odps_write_test00_partitioned", "partition": "school=SiChuan-School,class=1", "column": [ "id", "name" ], "accessId": "xxx", "accessKey": "xxxx", "truncate": true, "odpsServer": "http://sxxx/api", "tunnelServer": "http://xxx", "accountType": "aliyun" } } } ] } } 3.2 参数说明 accessId 描述：ODPS系统登录ID 必选：是 默认值：无 accessKey 描述：ODPS系统登录Key 必选：是 默认值：无 project 描述：ODPS表所属的project，注意:Project只能是字母+数字组合，请填写英文名称。在云端等用户看到的ODPS项目中文名只是显示名，请务必填写底层真实地Project英文标识名。 必选：是 默认值：无 table 描述：写入数据的表名，不能填写多张表，因为DataX不支持同时导入多张表。 必选：是 默认值：无 partition 描述：需要写入数据表的分区信息，必须指定到最后一级分区。把数据写入一个三级分区表，必须配置到最后一级分区，例如pt=20150101/type＝1/biz=2。 必选：如果是分区表，该选项必填，如果非分区表，该选项不可填写。 默认值：空 column 描述：需要导入的字段列表，当导入全部字段时，可以配置为"column": ["*"], 当需要插入部分odps列填写部分列，例如"column": [“id”, “name”]。ODPSWriter支持列筛选、列换序，例如表有a,b,c三个字段，用户只同步c,b两个字段。可以配置成[“c”,“b”], 在导入过程中，字段a自动补空，设置为null。 必选：否 默认值：无 truncate 描述：ODPSWriter通过配置"truncate": true，保证写入的幂等性，即当出现写入失败再次运行时，ODPSWriter将清理前述数据，并导入新数据，这样可以保证每次重跑之后的数据都保持一致。 **truncate选项不是原子操作！ODPS SQL无法做到原子性。因此当多个任务同时向一个Table/Partition清理分区时候，可能出现并发时序问题，请务必注意！**针对这类问题，我们建议尽量不要多个作业DDL同时操作同一份分区，或者在多个并发作业启动前，提前创建分区。 必选：是 默认值：无 odpsServer 描述：ODPS的server地址，线上地址为 http://service.odps.aliyun.com/api 必选：是 默认值：无 tunnelServer 描述：ODPS的tunnelserver地址，线上地址为 http://dt.odps.aliyun.com 必选：是， 默认值：无 ...

DataX OracleWriter 1 快速介绍 OracleWriter 插件实现了写入数据到 Oracle 主库的目的表的功能。在底层实现上， OracleWriter 通过 JDBC 连接远程 Oracle 数据库，并执行相应的 insert into … sql 语句将数据写入 Oracle，内部会分批次提交入库。 OracleWriter 面向ETL开发工程师，他们使用 OracleWriter 从数仓导入数据到 Oracle。同时 OracleWriter 亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 OracleWriter 通过 DataX 框架获取 Reader 生成的协议数据，根据你配置生成相应的SQL语句 insert into...(当主键/唯一性索引冲突时会写不进去冲突的行) 注意： 1. 目的表所在数据库必须是主库才能写入数据；整个任务至少需具备 insert into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 2.OracleWriter和MysqlWriter不同，不支持配置writeMode参数。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 Oracle 导入的数据。 { "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": "oraclewriter", "parameter": { "username": "root", "password": "root", "column": [ "id", "name" ], "preSql": [ "delete from test" ], "connection": [ { "jdbcUrl": "jdbc:oracle:thin:@[HOST_NAME]:PORT:[DATABASE_NAME]", "table": [ "test" ] } ] } } } ] } } 3.2 参数说明 jdbcUrl 描述：目的数据库的 JDBC 连接信息 ,jdbcUrl必须包含在connection配置单元中。 注意：1、在一个数据库上只能配置一个值。这与 OracleReader 支持多个备库探测不同，因为此处不支持同一个数据库存在多个主库的情况(双主导入数据情况) 2、jdbcUrl按照Oracle官方规范，并可以填写连接附加参数信息。具体请参看 Oracle官方文档或者咨询对应 DBA。 必选：是 默认值：无 username 描述：目的数据库的用户名 必选：是 默认值：无 password 描述：目的数据库的密码 必选：是 默认值：无 table 描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。 注意：table 和 jdbcUrl 必须包含在 connection 配置单元中 必选：是 默认值：无 column 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: “column”: [“id”,“name”,“age”]。如果要依次写入全部列，使用表示, 例如: “column”: [""] **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与Oracle的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。 必选：否 默认值：1024 session 描述：设置oracle连接时的session信息，格式示例如下： "session":[ "alter session set nls_date_format = 'dd.mm.yyyy hh24:mi:ss';" "alter session set NLS_LANG = 'AMERICAN';" ] 必选：否 默认值：无 ...

DataX OSSReader 说明 1 快速介绍 OSSReader提供了读取OSS数据存储的能力。在底层实现上，OSSReader使用OSS官方Java SDK获取OSS数据，并转换为DataX传输协议传递给Writer。 OSS 产品介绍, 参看[阿里云OSS Portal] OSS Java SDK, 参看[阿里云OSS Java SDK] 2 功能与限制 OSSReader实现了从OSS读取数据并转为DataX协议的功能，OSS本身是无结构化数据存储，对于DataX而言，OSSReader实现上类比TxtFileReader，有诸多相似之处。目前OSSReader支持功能如下： 支持且仅支持读取TXT的文件，且要求TXT中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 支持多种类型数据读取(使用String表示)，支持列裁剪，支持列常量 支持递归读取、支持文件名过滤。 支持文本压缩，现有压缩格式为zip、gzip、bzip2。注意，一个压缩包不允许多文件打包压缩。 多个object可以支持并发读取。 我们暂时不能做到： 单个Object(File)支持多线程并发读取，这里涉及到单个Object内部切分算法。二期考虑支持。 单个Object在压缩情况下，从技术上无法支持多线程并发读取。 3 功能说明 3.1 配置样例 { "job": { "setting": {}, "content": [ { "reader": { "name": "ossreader", "parameter": { "endpoint": "http://oss.aliyuncs.com", "accessId": "", "accessKey": "", "bucket": "myBucket", "object": [ "bazhen/*" ], "column": [ { "type": "long", "index": 0 }, { "type": "string", "value": "alibaba" }, { "type": "date", "index": 1, "format": "yyyy-MM-dd" } ], "encoding": "UTF-8", "fieldDelimiter": "\t", "compress": "gzip" } }, "writer": {} } ] } } 3.2 参数说明 endpoint 描述：OSS Server的EndPoint地址，例如http://oss.aliyuncs.com。 必选：是 默认值：无 accessId 描述：OSS的accessId 必选：是 默认值：无 accessKey 描述：OSS的accessKey 必选：是 默认值：无 bucket 描述：OSS的bucket 必选：是 默认值：无 object 描述：OSS的object信息，注意这里可以支持填写多个Object。 当指定单个OSS Object，OSSReader暂时只能使用单线程进行数据抽取。二期考虑在非压缩文件情况下针对单个Object可以进行多线程并发读取。 当指定多个OSS Object，OSSReader支持使用多线程进行数据抽取。线程并发数通过通道数指定。 当指定通配符，OSSReader尝试遍历出多个Object信息。例如: 指定/*代表读取bucket下游所有的Object，指定/bazhen/*代表读取bazhen目录下游所有的Object。 特别需要注意的是，DataX会将一个作业下同步的所有Object视作同一张数据表。用户必须自己保证所有的Object能够适配同一套schema信息。 必选：是 默认值：无 column 描述：读取字段列表，type指定源数据的类型，index指定当前列来自于文本第几列(以0开始)，value指定当前类型为常量，不从源头文件读取数据，而是根据value值自动生成对应的列。 默认情况下，用户可以全部按照String类型读取数据，配置如下： "column": ["*"] 用户可以指定Column字段信息，配置如下： { "type": "long", "index": 0 //从OSS文本第一列获取int字段 }, { "type": "string", "value": "alibaba" //从OSSReader内部生成alibaba的字符串字段作为当前字段 } 对于用户指定Column信息，type必须填写，index/value必须选择其一。 必选：是 默认值：全部按照string类型读取 fieldDelimiter 描述：读取的字段分隔符 必选：是 默认值：, compress 描述：文本压缩类型，默认不填写意味着没有压缩。支持压缩类型为zip、gzip、bzip2。 必选：否 默认值：不压缩 encoding 描述：读取文件的编码配置，目前只支持utf-8/gbk。 必选：否 默认值：utf-8 nullFormat 描述：文本文件中无法使用标准字符串定义null(空指针)，DataX提供nullFormat定义哪些字符串可以表示为null。 例如如果用户配置: nullFormat="\N"，那么如果源头数据是"\N"，DataX视作null字段。 必选：否 默认值：\N skipHeader 描述：类CSV格式文件可能存在表头为标题情况，需要跳过。默认不跳过。 必选：否 默认值：false csvReaderConfig 描述：读取CSV类型文件参数配置，Map类型。读取CSV类型文件使用的CsvReader进行读取，会有很多配置，不配置则使用默认值。 必选：否 默认值：无 ...

DataX OSSWriter 说明 1 快速介绍 OSSWriter提供了向OSS写入类CSV格式的一个或者多个表文件。 写入OSS内容存放的是一张逻辑意义上的二维表，例如CSV格式的文本信息。 OSS 产品介绍, 参看[阿里云OSS Portal] OSS Java SDK, 参看[阿里云OSS Java SDK] 2 功能与限制 OSSWriter实现了从DataX协议转为OSS中的TXT文件功能，OSS本身是无结构化数据存储，OSSWriter需要在如下几个方面增加: 支持且仅支持写入 TXT的文件，且要求TXT中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 暂时不支持文本压缩。 支持多线程写入，每个线程写入不同子文件。 文件支持滚动，当文件大于某个size值或者行数值，文件需要切换。 [暂不支持] 我们不能做到： 单个文件不能支持并发写入。 3 功能说明 3.1 配置样例 { "job": { "setting": {}, "content": [ { "reader": { }, "writer": { "name": "osswriter", "parameter": { "endpoint": "http://oss.aliyuncs.com", "accessId": "", "accessKey": "", "bucket": "myBucket", "object": "cdo/datax", "encoding": "UTF-8", "fieldDelimiter": ",", "writeMode": "truncate|append|nonConflict" } } } ] } } 3.2 参数说明 endpoint 描述：OSS Server的EndPoint地址，例如http://oss.aliyuncs.com。 必选：是 默认值：无 accessId 描述：OSS的accessId 必选：是 默认值：无 accessKey 描述：OSS的accessKey 必选：是 默认值：无 bucket 描述：OSS的bucket 必选：是 默认值：无 object 描述：OSSWriter写入的文件名，OSS使用文件名模拟目录的实现。 使用"object": “datax”，写入object以datax开头，后缀添加随机字符串。 使用"object": “cdo/datax”，写入的object以cdo/datax开头，后缀随机添加字符串，/作为OSS模拟目录的分隔符。 必选：是 默认值：无 writeMode 描述：OSSWriter写入前数据清理处理： truncate，写入前清理object名称前缀匹配的所有object。例如: “object”: “abc”，将清理所有abc开头的object。 append，写入前不做任何处理，DataX OSSWriter直接使用object名称写入，并使用随机UUID的后缀名来保证文件名不冲突。例如用户指定的object名为datax，实际写入为datax_xxxxxx_xxxx_xxxx nonConflict，如果指定路径出现前缀匹配的object，直接报错。例如: “object”: “abc”，如果存在abc123的object，将直接报错。 必选：是 默认值：无 fieldDelimiter 描述：读取的字段分隔符 必选：否 默认值：, encoding 描述：写出文件的编码配置。 必选：否 默认值：utf-8 nullFormat 描述：文本文件中无法使用标准字符串定义null(空指针)，DataX提供nullFormat定义哪些字符串可以表示为null。 例如如果用户配置: nullFormat="\N"，那么如果源头数据是"\N"，DataX视作null字段。 必选：否 默认值：\N dateFormat 描述：日期类型的数据序列化到object中时的格式，例如 “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 header 描述：Oss写出时的表头，示例[‘id’, ’name’, ‘age’]。 必选：否 默认值：无 maxFileSize 描述：Oss写出时单个Object文件的最大大小，默认为10000*10MB，类似log4j日志打印时根据日志文件大小轮转。OSS分块上传时，每个分块大小为10MB，每个OSS InitiateMultipartUploadRequest支持的分块最大数量为10000。轮转发生时，object名字规则是：在原有object前缀加UUID随机数的基础上，拼接_1,_2,_3等后缀。 必选：否 默认值：100000MB ...

DataX PostgresqlWriter 1 快速介绍 PostgresqlWriter插件实现了写入数据到 PostgreSQL主库目的表的功能。在底层实现上，PostgresqlWriter通过JDBC连接远程 PostgreSQL 数据库，并执行相应的 insert into … sql 语句将数据写入 PostgreSQL，内部会分批次提交入库。 PostgresqlWriter面向ETL开发工程师，他们使用PostgresqlWriter从数仓导入数据到PostgreSQL。同时 PostgresqlWriter亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 PostgresqlWriter通过 DataX 框架获取 Reader 生成的协议数据，根据你配置生成相应的SQL插入语句 insert into...(当主键/唯一性索引冲突时会写不进去冲突的行) 注意： 1. 目的表所在数据库必须是主库才能写入数据；整个任务至少需具备 insert into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 2. PostgresqlWriter和MysqlWriter不同，不支持配置writeMode参数。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 PostgresqlWriter导入的数据。 { "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": "postgresqlwriter", "parameter": { "username": "xx", "password": "xx", "column": [ "id", "name" ], "preSql": [ "delete from test" ], "connection": [ { "jdbcUrl": "jdbc:postgresql://127.0.0.1:3002/datax", "table": [ "test" ] } ] } } } ] } } 3.2 参数说明 jdbcUrl 描述：目的数据库的 JDBC 连接信息 ,jdbcUrl必须包含在connection配置单元中。 注意：1、在一个数据库上只能配置一个值。 2、jdbcUrl按照PostgreSQL官方规范，并可以填写连接附加参数信息。具体请参看PostgreSQL官方文档或者咨询对应 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与PostgreSql的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。 必选：否 默认值：1024 ...

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

DataX Transformer Transformer定义 在数据同步、传输过程中，存在用户对于数据传输进行特殊定制化的需求场景，包括裁剪列、转换列等工作，可以借助ETL的T过程实现(Transformer)。DataX包含了完整的E(Extract)、T(Transformer)、L(Load)支持。 运行模型 UDF手册 dx_substr 参数：3个 第一个参数：字段编号，对应record中第几个字段。 第二个参数：字段值的开始位置。 第三个参数：目标字段长度。 返回： 从字符串的指定位置（包含）截取指定长度的字符串。如果开始位置非法抛出异常。如果字段为空值，直接返回（即不参与本transformer） 举例： dx_substr(1,"2","5") column 1的value为“dataxTest”=>"taxTe" dx_substr(1,"5","10") column 1的value为“dataxTest”=>"Test" dx_pad 参数：4个 第一个参数：字段编号，对应record中第几个字段。 第二个参数：“l”,“r”, 指示是在头进行pad，还是尾进行pad。 第三个参数：目标字段长度。 第四个参数：需要pad的字符。 返回： 如果源字符串长度小于目标字段长度，按照位置添加pad字符后返回。如果长于，直接截断（都截右边）。如果字段为空值，转换为空字符串进行pad，即最后的字符串全是需要pad的字符 举例： dx_pad(1,"l","4","A"), 如果column 1 的值为 xyz=> Axyz， 值为 xyzzzzz => xyzz dx_pad(1,"r","4","A"), 如果column 1 的值为 xyz=> xyzA， 值为 xyzzzzz => xyzz dx_replace 参数：4个 第一个参数：字段编号，对应record中第几个字段。 第二个参数：字段值的开始位置。 第三个参数：需要替换的字段长度。 第四个参数：需要替换的字符串。 返回： 从字符串的指定位置（包含）替换指定长度的字符串。如果开始位置非法抛出异常。如果字段为空值，直接返回（即不参与本transformer） 举例： dx_replace(1,"2","4","****") column 1的value为“dataxTest”=>"da****est" dx_replace(1,"5","10","****") column 1的value为“dataxTest”=>"data****" dx_filter （关联filter暂不支持，即多个字段的联合判断，函参太过复杂，用户难以使用。） 参数： 第一个参数：字段编号，对应record中第几个字段。 第二个参数：运算符，支持一下运算符：like, not like, >, =, <, >=, !=, <= 第三个参数：正则表达式（java正则表达式）、值。 返回： 如果匹配正则表达式，返回Null，表示过滤该行。不匹配表达式时，表示保留该行。（注意是该行）。对于>=<都是对字段直接compare的结果. like ， not like是将字段转换成String，然后和目标正则表达式进行全匹配。 , =, <, >=, !=, <= 对于DoubleColumn比较double值，对于LongColumn和DateColumn比较long值，其他StringColumn，BooleanColumn以及ByteColumn均比较的是StringColumn值。 如果目标colunn为空（null），对于 = null的过滤条件，将满足条件，被过滤。！=null的过滤条件，null不满足过滤条件，不被过滤。 like，字段为null不满足条件，不被过滤，和not like，字段为null满足条件，被过滤。 举例： dx_filter(1,"like","dataTest") dx_filter(1,">=","10") dx_groovy 参数。 第一个参数： groovy code 第二个参数（列表或者为空）：extraPackage 备注： dx_groovy只能调用一次。不能多次调用。 groovy code中支持java.lang, java.util的包，可直接引用的对象有record，以及element下的各种column（BoolColumn.class,BytesColumn.class,DateColumn.class,DoubleColumn.class,LongColumn.class,StringColumn.class）。不支持其他包，如果用户有需要用到其他包，可设置extraPackage，注意extraPackage不支持第三方jar包。 groovy code中，返回更新过的Record（比如record.setColumn(columnIndex, new StringColumn(newValue));），或者null。返回null表示过滤此行。 用户可以直接调用静态的Util方式（GroovyTransformerStaticUtil），目前GroovyTransformerStaticUtil的方法列表 (按需补充)： 举例: groovy 实现的subStr: String code = "Column column = record.getColumn(1);\n" + " String oriValue = column.asString();\n" + " String newValue = oriValue.substring(0, 3);\n" + " record.setColumn(1, new StringColumn(newValue));\n" + " return record;"; dx_groovy(record); groovy 实现的Replace String code2 = "Column column = record.getColumn(1);\n" + " String oriValue = column.asString();\n" + " String newValue = \"****\" + oriValue.substring(3, oriValue.length());\n" + " record.setColumn(1, new StringColumn(newValue));\n" + " return record;"; groovy 实现的Pad String code3 = "Column column = record.getColumn(1);\n" + " String oriValue = column.asString();\n" + " String padString = \"12345\";\n" + " String finalPad = \"\";\n" + " int NeedLength = 8 - oriValue.length();\n" + " while (NeedLength > 0) {\n" + "\n" + " if (NeedLength >= padString.length()) {\n" + " finalPad += padString;\n" + " NeedLength -= padString.length();\n" + " } else {\n" + " finalPad += padString.substring(0, NeedLength);\n" + " NeedLength = 0;\n" + " }\n" + " }\n" + " String newValue= finalPad + oriValue;\n" + " record.setColumn(1, new StringColumn(newValue));\n" + " return record;"; Job定义 本例中，配置3个UDF。 { "job": { "setting": { "speed": { "channel": 1 }, "errorLimit": { "record": 0 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ { "value": "DataX", "type": "string" }, { "value": 19890604, "type": "long" }, { "value": "1989-06-04 00:00:00", "type": "date" }, { "value": true, "type": "bool" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 100000 } }, "writer": { "name": "streamwriter", "parameter": { "print": false, "encoding": "UTF-8" } }, "transformer": [ { "name": "dx_substr", "parameter": { "columnIndex":5, "paras":["1","3"] } }, { "name": "dx_replace", "parameter": { "columnIndex":4, "paras":["3","4","****"] } }, { "name": "dx_groovy", "parameter": { "code": "//groovy code//", "extraPackage":[ "import somePackage1;", "import somePackage2;" ] } } ] } ] } } 计量和脏数据 Transform过程涉及到数据的转换，可能造成数据的增加或减少，因此更加需要精确度量，包括： ...

DataX TxtFileReader 说明 1 快速介绍 TxtFileReader提供了读取本地文件系统数据存储的能力。在底层实现上，TxtFileReader获取本地文件数据，并转换为DataX传输协议传递给Writer。 本地文件内容存放的是一张逻辑意义上的二维表，例如CSV格式的文本信息。 2 功能与限制 TxtFileReader实现了从本地文件读取数据并转为DataX协议的功能，本地文件本身是无结构化数据存储，对于DataX而言，TxtFileReader实现上类比OSSReader，有诸多相似之处。目前TxtFileReader支持功能如下： 支持且仅支持读取TXT的文件，且要求TXT中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 支持多种类型数据读取(使用String表示)，支持列裁剪，支持列常量 支持递归读取、支持文件名过滤。 支持文本压缩，现有压缩格式为zip、gzip、bzip2。 多个File可以支持并发读取。 我们暂时不能做到： 单个File支持多线程并发读取，这里涉及到单个File内部切分算法。二期考虑支持。 单个File在压缩情况下，从技术上无法支持多线程并发读取。 3 功能说明 3.1 配置样例 { "setting": {}, "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "txtfilereader", "parameter": { "path": ["/home/haiwei.luo/case00/data"], "encoding": "UTF-8", "column": [ { "index": 0, "type": "long" }, { "index": 1, "type": "boolean" }, { "index": 2, "type": "double" }, { "index": 3, "type": "string" }, { "index": 4, "type": "date", "format": "yyyy.MM.dd" } ], "fieldDelimiter": "," } }, "writer": { "name": "txtfilewriter", "parameter": { "path": "/home/haiwei.luo/case00/result", "fileName": "luohw", "writeMode": "truncate", "format": "yyyy-MM-dd" } } } ] } } 3.2 参数说明 path 描述：本地文件系统的路径信息，注意这里可以支持填写多个路径。 当指定单个本地文件，TxtFileReader暂时只能使用单线程进行数据抽取。二期考虑在非压缩文件情况下针对单个File可以进行多线程并发读取。 当指定多个本地文件，TxtFileReader支持使用多线程进行数据抽取。线程并发数通过通道数指定。 当指定通配符，TxtFileReader尝试遍历出多个文件信息。例如: 指定/*代表读取/目录下所有的文件，指定/bazhen/*代表读取bazhen目录下游所有的文件。TxtFileReader目前只支持*作为文件通配符。 特别需要注意的是，DataX会将一个作业下同步的所有Text File视作同一张数据表。用户必须自己保证所有的File能够适配同一套schema信息。读取文件用户必须保证为类CSV格式，并且提供给DataX权限可读。 特别需要注意的是，如果Path指定的路径下没有符合匹配的文件抽取，DataX将报错。 必选：是 默认值：无 column 描述：读取字段列表，type指定源数据的类型，index指定当前列来自于文本第几列(以0开始)，value指定当前类型为常量，不从源头文件读取数据，而是根据value值自动生成对应的列。 默认情况下，用户可以全部按照String类型读取数据，配置如下： "column": ["*"] 用户可以指定Column字段信息，配置如下： { "type": "long", "index": 0 //从本地文件文本第一列获取int字段 }, { "type": "string", "value": "alibaba" //从TxtFileReader内部生成alibaba的字符串字段作为当前字段 } 对于用户指定Column信息，type必须填写，index/value必须选择其一。 必选：是 默认值：全部按照string类型读取 fieldDelimiter 描述：读取的字段分隔符 必选：是 默认值：, compress 描述：文本压缩类型，默认不填写意味着没有压缩。支持压缩类型为zip、gzip、bzip2。 必选：否 默认值：没有压缩 encoding 描述：读取文件的编码配置。 必选：否 默认值：utf-8 skipHeader 描述：类CSV格式文件可能存在表头为标题情况，需要跳过。默认不跳过。 必选：否 默认值：false nullFormat 描述：文本文件中无法使用标准字符串定义null(空指针)，DataX提供nullFormat定义哪些字符串可以表示为null。 例如如果用户配置: nullFormat:"\N"，那么如果源头数据是"\N"，DataX视作null字段。 必选：否 默认值：\N csvReaderConfig 描述：读取CSV类型文件参数配置，Map类型。读取CSV类型文件使用的CsvReader进行读取，会有很多配置，不配置则使用默认值。 必选：否 默认值：无 ...

DataX TxtFileWriter 说明 1 快速介绍 TxtFileWriter提供了向本地文件写入类CSV格式的一个或者多个表文件。TxtFileWriter服务的用户主要在于DataX开发、测试同学。 写入本地文件内容存放的是一张逻辑意义上的二维表，例如CSV格式的文本信息。 2 功能与限制 TxtFileWriter实现了从DataX协议转为本地TXT文件功能，本地文件本身是无结构化数据存储，TxtFileWriter如下几个方面约定: 支持且仅支持写入 TXT的文件，且要求TXT中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 支持文本压缩，现有压缩格式为gzip、bzip2。 支持多线程写入，每个线程写入不同子文件。 文件支持滚动，当文件大于某个size值或者行数值，文件需要切换。 [暂不支持] 我们不能做到： 单个文件不能支持并发写入。 3 功能说明 3.1 配置样例 { "setting": {}, "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "txtfilereader", "parameter": { "path": ["/home/haiwei.luo/case00/data"], "encoding": "UTF-8", "column": [ { "index": 0, "type": "long" }, { "index": 1, "type": "boolean" }, { "index": 2, "type": "double" }, { "index": 3, "type": "string" }, { "index": 4, "type": "date", "format": "yyyy.MM.dd" } ], "fieldDelimiter": "," } }, "writer": { "name": "txtfilewriter", "parameter": { "path": "/home/haiwei.luo/case00/result", "fileName": "luohw", "writeMode": "truncate", "dateFormat": "yyyy-MM-dd" } } } ] } } 3.2 参数说明 path 描述：本地文件系统的路径信息，TxtFileWriter会写入Path目录下属多个文件。 必选：是 默认值：无 fileName 描述：TxtFileWriter写入的文件名，该文件名会添加随机的后缀作为每个线程写入实际文件名。 必选：是 默认值：无 writeMode 描述：TxtFileWriter写入前数据清理处理模式： truncate，写入前清理目录下一fileName前缀的所有文件。 append，写入前不做任何处理，DataX TxtFileWriter直接使用filename写入，并保证文件名不冲突。 nonConflict，如果目录下有fileName前缀的文件，直接报错。 必选：是 默认值：无 fieldDelimiter 描述：读取的字段分隔符 必选：否 默认值：, compress 描述：文本压缩类型，默认不填写意味着没有压缩。支持压缩类型为zip、lzo、lzop、tgz、bzip2。 必选：否 默认值：无压缩 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 header 描述：txt写出时的表头，示例[‘id’, ’name’, ‘age’]。 必选：否 默认值：无 ...