Datax

RDBMSWriter 插件文档 1 快速介绍 RDBMSWriter 插件实现了写入数据到 RDBMS 主库的目的表的功能。在底层实现上， RDBMSWriter 通过 JDBC 连接远程 RDBMS 数据库，并执行相应的 insert into … 的 sql 语句将数据写入 RDBMS。 RDBMSWriter是一个通用的关系数据库写插件，您可以通过注册数据库驱动等方式增加任意多样的关系数据库写支持。 RDBMSWriter 面向ETL开发工程师，他们使用 RDBMSWriter 从数仓导入数据到 RDBMS。同时 RDBMSWriter 亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 RDBMSWriter 通过 DataX 框架获取 Reader 生成的协议数据，RDBMSWriter 通过 JDBC 连接远程 RDBMS 数据库，并执行相应的 insert into … 的 sql 语句将数据写入 RDBMS。 3 功能说明 3.1 配置样例 配置一个写入RDBMS的作业。 { "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": "rdbmswriter", "parameter": { "connection": [ { "jdbcUrl": "jdbc:dm://ip:port/database", "table": [ "table" ] } ], "username": "username", "password": "password", "table": "table", "column": [ "*" ], "preSql": [ "delete from XXX;" ] } } } ] } } 3.2 参数说明 jdbcUrl 描述：描述的是到对端数据库的JDBC连接信息，jdbcUrl按照RDBMS官方规范，并可以填写连接附件控制信息。请注意不同的数据库jdbc的格式是不同的，DataX会根据具体jdbc的格式选择合适的数据库驱动完成数据读取。 达梦 jdbc:dm://ip:port/database db2格式 jdbc:db2://ip:port/database PPAS格式 jdbc:edb://ip:port/database rdbmswriter如何增加新的数据库支持: 进入rdbmswriter对应目录，这里${DATAX_HOME}为DataX主目录，即: ${DATAX_HOME}/plugin/writer/rdbmswriter 在rdbmswriter插件目录下有plugin.json配置文件，在此文件中注册您具体的数据库驱动，具体放在drivers数组中。rdbmswriter插件在任务执行时会动态选择合适的数据库驱动连接数据库。 { "name": "rdbmswriter", "class": "com.alibaba.datax.plugin.reader.rdbmswriter.RdbmsWriter", "description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.", "developer": "alibaba", "drivers": [ "dm.jdbc.driver.DmDriver", "com.ibm.db2.jcc.DB2Driver", "com.sybase.jdbc3.jdbc.SybDriver", "com.edb.Driver" ] } 在rdbmswriter插件目录下有libs子目录，您需要将您具体的数据库驱动放到libs目录下。 $tree . |-- libs | |-- Dm7JdbcDriver16.jar | |-- commons-collections-3.0.jar | |-- commons-io-2.4.jar | |-- commons-lang3-3.3.2.jar | |-- commons-math3-3.1.1.jar | |-- datax-common-0.0.1-SNAPSHOT.jar | |-- datax-service-face-1.0.23-20160120.024328-1.jar | |-- db2jcc4.jar | |-- druid-1.0.15.jar | |-- edb-jdbc16.jar | |-- fastjson-1.1.46.sec01.jar | |-- guava-r05.jar | |-- hamcrest-core-1.3.jar | |-- jconn3-1.0.0-SNAPSHOT.jar | |-- logback-classic-1.0.13.jar | |-- logback-core-1.0.13.jar | |-- plugin-rdbms-util-0.0.1-SNAPSHOT.jar | `-- slf4j-api-1.7.10.jar |-- plugin.json |-- plugin_job_template.json `-- rdbmswriter-0.0.1-SNAPSHOT.jar 必选：是 默认值：无 username 描述：数据源的用户名 必选：是 默认值：无 password 描述：数据源指定用户名的密码 必选：是 默认值：无 table 描述：目标表名称，如果表的schema信息和上述配置username不一致，请使用schema.table的格式填写table信息。 必选：是 默认值：无 column 描述：所配置的表中需要同步的列名集合。以英文逗号（,）进行分隔。我们强烈不推荐用户使用默认列情况 必选：是 默认值：无 preSql 描述：执行数据同步任务之前率先执行的sql语句，目前只允许执行一条SQL语句，例如清除旧数据。 必选：否 默认值：无 postSql 描述：执行数据同步任务之后执行的sql语句，目前只允许执行一条SQL语句，例如加上某一个时间戳。 必选：否 默认值：无 batchSize 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与RDBMS的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。 必选：否 默认值：1024 ...

some script here.

本插件仅在Elasticsearch 5.x上测试

SqlServerReader 插件文档 1 快速介绍 SqlServerReader插件实现了从SqlServer读取数据。在底层实现上，SqlServerReader通过JDBC连接远程SqlServer数据库，并执行相应的sql语句将数据从SqlServer库中SELECT出来。 2 实现原理 简而言之，SqlServerReader通过JDBC连接器连接到远程的SqlServer数据库，并根据用户配置的信息生成查询SELECT SQL语句并发送到远程SqlServer数据库，并将该SQL执行返回结果使用DataX自定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。 对于用户配置Table、Column、Where的信息，SqlServerReader将其拼接为SQL语句发送到SqlServer数据库；对于用户配置querySql信息，SqlServer直接将其发送到SqlServer数据库。 3 功能说明 3.1 配置样例 配置一个从SqlServer数据库同步抽取数据到本地的作业: { "job": { "setting": { "speed": { "byte": 1048576 } }, "content": [ { "reader": { "name": "sqlserverreader", "parameter": { // 数据库连接用户名 "username": "root", // 数据库连接密码 "password": "root", "column": [ "id" ], "splitPk": "db_id", "connection": [ { "table": [ "table" ], "jdbcUrl": [ "jdbc:sqlserver://localhost:3433;DatabaseName=dbname" ] } ] } }, "writer": { "name": "streamwriter", "parameter": { "print": true, "encoding": "UTF-8" } } } ] } } 配置一个自定义SQL的数据库同步任务到本地内容的作业： { "job": { "setting": { "speed": 1048576 }, "content": [ { "reader": { "name": "sqlserverreader", "parameter": { "username": "root", "password": "root", "where": "", "connection": [ { "querySql": [ "select db_id,on_line_flag from db_info where db_id < 10;" ], "jdbcUrl": [ "jdbc:sqlserver://bad_ip:3433;DatabaseName=dbname", "jdbc:sqlserver://127.0.0.1:bad_port;DatabaseName=dbname", "jdbc:sqlserver://127.0.0.1:3306;DatabaseName=dbname" ] } ] } }, "writer": { "name": "streamwriter", "parameter": { "visible": false, "encoding": "UTF-8" } } } ] } } 3.2 参数说明 jdbcUrl 描述：描述的是到对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，SqlServerReader可以依次探测ip的可连接性，直到选择一个合法的IP。如果全部连接失败，SqlServerReader报错。 注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。 jdbcUrl按照SqlServer官方规范，并可以填写连接附件控制信息。具体请参看SqlServer官方文档。 必选：是 默认值：无 username 描述：数据源的用户名 必选：是 默认值：无 password 描述：数据源指定用户名的密码 必选：是 默认值：无 table 描述：所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema结构，SqlServerReader不予检查表是否同一逻辑表。注意，table必须包含在connection配置单元中。 必选：是 默认值：无 column 描述：所配置的表中需要同步的列名集合，使用JSON的数组描述字段信息。用户使用*代表默认使用所有列配置，例如["*"]。 支持列裁剪，即列可以挑选部分列进行导出。 支持列换序，即列可以不按照表schema信息进行导出。 支持常量配置，用户需要按照JSON格式: [“id”, “[table]”, “1”, “‘bazhen.csy’”, “null”, “COUNT(*)”, “2.3” , “true”] id为普通列名，[table]为包含保留在的列名，1为整形数字常量，‘bazhen.csy’为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。 column必须用户显示指定同步的列集合，不允许为空！ 必选：是 默认值：无 splitPk 描述：SqlServerReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能。 推荐splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。 目前splitPk仅支持整形型数据切分，不支持浮点、字符串、日期等其他类型。如果用户指定其他非支持类型，SqlServerReader将报错！ splitPk设置为空，底层将视作用户不允许对单表进行切分，因此使用单通道进行抽取。 必选：否 默认值：无 where 描述：筛选条件，MysqlReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。 where条件可以有效地进行业务增量同步。如果该值为空，代表同步全表所有的信息。 必选：否 默认值：无 querySql 描述：在有些业务场景下，where这一配置项不足以描述所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX系统就会忽略table，column这些配置型，直接使用这个配置项的内容对数据进行筛选，例如需要进行多表join后同步数据，使用select a,b from table_a join table_b on table_a.id = table_b.id 当用户配置querySql时，SqlServerReader直接忽略table、column、where条件的配置。 必选：否 默认值：无 fetchSize 描述：该配置项定义了插件和数据库服务器端每次批量数据获取条数，该值决定了DataX和服务器端的网络交互次数，能够较大的提升数据抽取性能。 注意，该值过大(>2048)可能造成DataX进程OOM。。 必选：否 默认值：1024 ...

TableStore增量数据导出通道：TableStoreStreamReader 快速介绍 TableStoreStreamReader插件主要用于TableStore的增量数据导出，增量数据可以看作操作日志，除了数据本身外还附有操作信息。 与全量导出插件不同，增量导出插件只有多版本模式，同时不支持指定列。这是与增量导出的原理有关的，导出的格式下面有详细介绍。 使用插件前必须确保表上已经开启Stream功能，可以在建表的时候指定开启，或者使用SDK的UpdateTable接口开启。 开启Stream的方法： SyncClient client = new SyncClient("", "", "", ""); 1. 建表的时候开启： CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta); createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // 24代表增量数据保留24小时 client.createTable(createTableRequest); 2. 如果建表时未开启，可以通过UpdateTable开启: UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName"); updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); client.updateTable(updateTableRequest); 实现原理 首先用户使用SDK的UpdateTable功能，指定开启Stream并设置过期时间，即开启了增量功能。 开启后，TableStore服务端就会将用户的操作日志额外保存起来， 每个分区有一个有序的操作日志队列，每条操作日志会在一定时间后被垃圾回收，这个时间即用户指定的过期时间。 TableStore的SDK提供了几个Stream相关的API用于将这部分操作日志读取出来，增量插件也是通过TableStore SDK的接口获取到增量数据的，并将 增量数据转化为多个6元组的形式(pk, colName, version, colValue, opType, sequenceInfo)导入到ODPS中。 Reader的配置模版： "reader": { "name" : "otsstreamreader", "parameter" : { "endpoint" : "", "accessId" : "", "accessKey" : "", "instanceName" : "", //dataTable即需要导出数据的表。 "dataTable" : "", //statusTable是Reader用于保存状态的表，若该表不存在，Reader会自动创建该表。 //一次离线导出任务完成后，用户不应删除该表，该表中记录的状态可用于下次导出任务中。 "statusTable" : "TableStoreStreamReaderStatusTable", //增量数据的时间范围（左闭右开）的左边界。 "startTimestampMillis" : "", //增量数据的时间范围（左闭右开）的右边界。 "endTimestampMillis" : "", //采云间调度只支持天级别，所以提供该配置，作用与startTimestampMillis和endTimestampMillis类似。 "date": "", //是否导出时序信息。 "isExportSequenceInfo": true, //从TableStore中读增量数据时，每次请求的最大重试次数，默认为30。 "maxRetries" : 30 } } 参数说明 名称 说明 类型 必选 endpoint TableStoreServer的Endpoint地址。 String 是 accessId 用于访问TableStore服务的accessId。 String 是 accessKey 用于访问TableStore服务的accessKey。 String 是 instanceName TableStore的实例名称。 String 是 dataTable 需要导出增量数据的表的名称。该表需要开启Stream，可以在建表时开启，或者使用UpdateTable接口开启。 String 是 statusTable Reader插件用于记录状态的表的名称，这些状态可用于减少对非目标范围内的数据的扫描，从而加快导出速度。 1. 用户不需要创建该表，只需要给出一个表名。Reader插件会尝试在用户的instance下创建该表，若该表不存在即创建新表，若该表已存在，会判断该表的Meta是否与期望一致，若不一致会抛出异常。 2. 在一次导出完成之后，用户不应删除该表，该表的状态可用于下次导出任务。 3. 该表会开启TTL，数据自动过期，因此可认为其数据量很小。 4. 针对同一个instance下的多个不同的dataTable的Reader配置，可以使用同一个statusTable，记录的状态信息互不影响。 综上，用户配置一个类似TableStoreStreamReaderStatusTable之类的名称即可，注意不要与业务相关的表重名。 String 是 startTimestampMillis 增量数据的时间范围（左闭右开）的左边界，单位毫秒。 1. Reader插件会从statusTable中找对应startTimestampMillis的位点，从该点开始读取开始导出数据。 2. 若statusTable中找不到对应的位点，则从系统保留的增量数据的第一条开始读取，并跳过写入时间小于startTimestampMillis的数据。 Long 否 endTimestampMillis 增量数据的时间范围（左闭右开）的右边界，单位毫秒。 1. Reader插件从startTimestampMillis位置开始导出数据后，当遇到第一条时间戳大于等于endTimestampMillis的数据时，结束导出数据，导出完成。 2. 当读取完当前全部的增量数据时，结束读取，即使未达到endTimestampMillis。 Long 否 date 日期格式为yyyyMMdd，如20151111，表示导出该日的数据。 若没有指定date，则必须指定startTimestampMillis和endTimestampMillis，反之也成立。 String 否 isExportSequenceInfo 是否导出时序信息，时序信息包含了数据的写入时间等。默认该值为false，即不导出。 Boolean 否 maxRetries 从TableStore中读增量数据时，每次请求的最大重试次数，默认为30，重试之间有间隔，30次重试总时间约为5分钟，一般无需更改。 Int 否 导出的数据格式 首先，在TableStore多版本模型下，表中的数据组织为“行－列－版本”三级的模式， 一行可以有任意列，列名也并非固定的，每一列可以含有多个版本，每个版本都有一个特定的时间戳（版本号）。 ...

TSDBReader 插件文档 1 快速介绍 TSDBReader 插件实现了从阿里云 TSDB 读取数据。阿里云时间序列数据库 ( Time Series Database , 简称 TSDB) 是一种集时序数据高效读写，压缩存储，实时计算能力为一体的数据库服务，可广泛应用于物联网和互联网领域，实现对设备及业务服务的实时监控，实时预测告警。详见 TSDB 的阿里云官网。 2 实现原理 在底层实现上，TSDBReader 通过 HTTP 请求链接到 阿里云 TSDB 实例，利用 /api/query 或者 /api/mquery 接口将数据点扫描出来（更多细节详见：时序数据库 TSDB - HTTP API 概览）。而整个同步的过程，是通过时间线和查询时间线范围进行切分。 3 功能说明 3.1 配置样例 配置一个从 阿里云 TSDB 数据库同步抽取数据到本地的作业，并以时序数据的格式输出： 时序数据样例： {"metric":"m","tags":{"app":"a19","cluster":"c5","group":"g10","ip":"i999","zone":"z1"},"timestamp":1546272263,"value":1} { "job": { "content": [ { "reader": { "name": "tsdbreader", "parameter": { "sinkDbType": "TSDB", "endpoint": "http://localhost:8242", "column": [ "m" ], "splitIntervalMs": 60000, "beginDateTime": "2019-01-01 00:00:00", "endDateTime": "2019-01-01 01:00:00" } }, "writer": { "name": "streamwriter", "parameter": { "encoding": "UTF-8", "print": true } } } ], "setting": { "speed": { "channel": 3 } } } } 配置一个从 阿里云 TSDB 数据库同步抽取数据到本地的作业，并以关系型数据的格式输出： 关系型数据样例： ...

TSDBWriter 插件文档 1 快速介绍 TSDBWriter 插件实现了将数据点写入到阿里巴巴自研 TSDB 数据库中（后续简称 TSDB）。 时间序列数据库（Time Series Database , 简称 TSDB）是一种高性能，低成本，稳定可靠的在线时序数据库服务；提供高效读写，高压缩比存储、时序数据插值及聚合计算，广泛应用于物联网（IoT）设备监控系统 ，企业能源管理系统（EMS），生产安全监控系统，电力检测系统等行业场景。 TSDB 提供百万级时序数据秒级写入，高压缩比低成本存储、预降采样、插值、多维聚合计算，查询结果可视化功能；解决由于设备采集点数量巨大，数据采集频率高，造成的存储成本高，写入和查询分析效率低的问题。更多关于 TSDB 的介绍，详见阿里云 TSDB 官网。 2 实现原理 通过 HTTP 连接 TSDB 实例，并通过 /api/put 接口将数据点写入。关于写入接口详见 TSDB 的接口说明文档。 3 功能说明 3.1 配置样例 配置一个从 OpenTSDB 数据库同步抽取数据到 TSDB： { "job": { "content": [ { "reader": { "name": "opentsdbreader", "parameter": { "endpoint": "http://localhost:4242", "column": [ "m" ], "startTime": "2019-01-01 00:00:00", "endTime": "2019-01-01 03:00:00" } }, "writer": { "name": "tsdbhttpwriter", "parameter": { "endpoint": "http://localhost:8242" } } } ], "setting": { "speed": { "channel": 1 } } } } 3.2 参数说明 name 描述：本插件的名称 必选：是 默认值：tsdbhttpwriter parameter endpoint 描述：TSDB 的 HTTP 连接地址 必选：是 格式：http://IP:Port 默认值：无 batchSize 描述：每次批量数据的条数 必选：否 格式：int，需要保证大于 0 默认值：100 maxRetryTime 描述：失败后重试的次数 必选：否 格式：int，需要保证大于 1 默认值：3 ignoreWriteError 描述：如果设置为 true，则忽略写入错误，继续写入；否则，多次重试后仍写入失败的话，则会终止写入任务 必选：否 格式：bool 默认值：false ...

阿里云开源离线同步工具DataX3.0介绍 一. DataX3.0概览 ​ DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。 设计理念 为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。 当前使用现状 DataX在阿里巴巴集团内被广泛使用，承担了所有大数据的离线同步业务，并已持续稳定运行了6年之久。目前每天完成同步8w多道作业，每日传输数据量超过300TB。 此前已经开源DataX1.0版本，此次介绍为阿里云开源全新版本DataX3.0，有了更多更强大的功能和更好的使用体验。Github主页地址：https://github.com/alibaba/DataX 二、DataX3.0框架设计 DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。 Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。 Writer： Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。 Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。 三. DataX3.0插件体系 ​ 经过几年积累，DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入。DataX目前支持数据如下： 类型 数据源 Reader(读) Writer(写) 文档 RDBMS 关系型数据库 MySQL √ √ 读 、写 Oracle √ √ 读 、写 SQLServer √ √ 读 、写 PostgreSQL √ √ 读 、写 DRDS √ √ 读 、写 达梦 √ √ 读 、写 通用RDBMS(支持所有关系型数据库) √ √ 读 、写 阿里云数仓数据存储 ODPS √ √ 读 、写 ADS √ 写 OSS √ √ 读 、写 OCS √ √ 读 、写 NoSQL数据存储 OTS √ √ 读 、写 Hbase0.94 √ √ 读 、写 Hbase1.1 √ √ 读 、写 MongoDB √ √ 读 、写 Hive √ √ 读 、写 无结构化数据存储 TxtFile √ √ 读 、写 FTP √ √ 读 、写 HDFS √ √ 读 、写 Elasticsearch √ 写 DataX Framework提供了简单的接口与插件交互，提供简单的插件接入机制，只需要任意加上一种插件，就能无缝对接其他数据源。详情请看：DataX数据源指南 ...