Datax

CassandraReader 插件文档 1 快速介绍 CassandraReader插件实现了从Cassandra读取数据。在底层实现上，CassandraReader通过datastax的java driver连接Cassandra实例，并执行相应的cql语句将数据从cassandra中SELECT出来。 2 实现原理 简而言之，CassandraReader通过java driver连接到Cassandra实例，并根据用户配置的信息生成查询SELECT CQL语句，然后发送到Cassandra，并将该CQL执行返回结果使用DataX自定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。 对于用户配置Table、Column的信息，CassandraReader将其拼接为CQL语句发送到Cassandra。 3 功能说明 3.1 配置样例 配置一个从Cassandra同步抽取数据到本地的作业: { "job": { "setting": { "speed": { "channel": 3 } }, "content": [ { "reader": { "name": "cassandrareader", "parameter": { "host": "localhost", "port": 9042, "useSSL": false, "keyspace": "test", "table": "datax_src", "column": [ "textCol", "blobCol", "writetime(blobCol)", "boolCol", "smallintCol", "tinyintCol", "intCol", "bigintCol", "varintCol", "floatCol", "doubleCol", "decimalCol", "dateCol", "timeCol", "timeStampCol", "uuidCol", "inetCol", "durationCol", "listCol", "mapCol", "setCol" "tupleCol" "udtCol", ] } }, "writer": { "name": "streamwriter", "parameter": { "print":true } } } ] } } 3.2 参数说明 host 描述：Cassandra连接点的域名或ip，多个node之间用逗号分隔。 必选：是 默认值：无 port 描述：Cassandra端口。 必选：是 默认值：9042 username 描述：数据源的用户名 必选：否 默认值：无 password 描述：数据源指定用户名的密码 必选：否 默认值：无 useSSL 描述：是否使用SSL连接。 必选：否 默认值：false keyspace 描述：需要同步的表所在的keyspace。 必选：是 默认值：无 table 描述：所选取的需要同步的表。 必选：是 默认值：无 column 描述：所配置的表中需要同步的列集合。 其中的元素可以指定列的名称或writetime(column_name)，后一种形式会读取column_name列的时间戳而不是数据。 必选：是 默认值：无 where 描述：数据筛选条件的cql表达式，例如: "where":"textcol='a'" 必选：否 默认值：无 allowFiltering 描述：是否在服务端过滤数据。参考cassandra文档中ALLOW FILTERING关键字的相关描述。 必选：否 默认值：无 consistancyLevel 描述：数据一致性级别。可选ONE|QUORUM|LOCAL_QUORUM|EACH_QUORUM|ALL|ANY|TWO|THREE|LOCAL_ONE 必选：否 默认值：LOCAL_QUORUM ...

CassandraWriter 插件文档 1 快速介绍 CassandraWriter插件实现了向Cassandra写入数据。在底层实现上，CassandraWriter通过datastax的java driver连接Cassandra实例，并执行相应的cql语句将数据写入cassandra中。 2 实现原理 简而言之，CassandraWriter通过java driver连接到Cassandra实例，并根据用户配置的信息生成INSERT CQL语句，然后发送到Cassandra。 对于用户配置Table、Column的信息，CassandraReader将其拼接为CQL语句发送到Cassandra。 3 功能说明 3.1 配置样例 配置一个从内存产生到Cassandra导入的作业: { "job": { "setting": { "speed": { "channel": 5 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ {"value":"name","type": "string"}, {"value":"false","type":"bool"}, {"value":"1988-08-08 08:08:08","type":"date"}, {"value":"addr","type":"bytes"}, {"value":1.234,"type":"double"}, {"value":12345678,"type":"long"}, {"value":2.345,"type":"double"}, {"value":3456789,"type":"long"}, {"value":"4a0ef8c0-4d97-11d0-db82-ebecdb03ffa5","type":"string"}, {"value":"value","type":"bytes"}, {"value":"-838383838,37377373,-383883838,27272772,393993939,-38383883,83883838,-1350403181,817650816,1630642337,251398784,-622020148","type":"string"}, ], "sliceRecordCount": 10000000 } }, "writer": { "name": "cassandrawriter", "parameter": { "host": "localhost", "port": 9042, "useSSL": false, "keyspace": "stresscql", "table": "dst", "batchSize":10, "column": [ "name", "choice", "date", "address", "dbl", "lval", "fval", "ival", "uid", "value", "listval" ] } } } ] } } 3.2 参数说明 host 描述：Cassandra连接点的域名或ip，多个node之间用逗号分隔。 必选：是 默认值：无 port 描述：Cassandra端口。 必选：是 默认值：9042 username 描述：数据源的用户名 必选：否 默认值：无 password 描述：数据源指定用户名的密码 必选：否 默认值：无 useSSL 描述：是否使用SSL连接。 必选：否 默认值：false connectionsPerHost 描述：客户端连接池配置：与服务器每个节点建多少个连接。 必选：否 默认值：8 maxPendingPerConnection 描述：客户端连接池配置：每个连接最大请求数。 必选：否 默认值：128 keyspace 描述：需要同步的表所在的keyspace。 必选：是 默认值：无 table 描述：所选取的需要同步的表。 必选：是 默认值：无 column 描述：所配置的表中需要同步的列集合。 内容可以是列的名称或"writetime()"。如果将列名配置为writetime()，会将这一列的内容作为时间戳。 必选：是 默认值：无 consistancyLevel 描述：数据一致性级别。可选ONE|QUORUM|LOCAL_QUORUM|EACH_QUORUM|ALL|ANY|TWO|THREE|LOCAL_ONE 必选：否 默认值：LOCAL_QUORUM batchSize 描述：一次批量提交(UNLOGGED BATCH)的记录数大小（条数）。注意batch的大小有如下限制： （1）不能超过65535。 (2) batch中的内容大小受到服务器端batch_size_fail_threshold_in_kb的限制。 (3) 如果batch中的内容超过了batch_size_warn_threshold_in_kb的限制，会打出warn日志，但并不影响写入，忽略即可。 如果批量提交失败，会把这个批量的所有内容重新逐条写入一遍。 必选：否 默认值：1 ...

DataX DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。 DataX 商业版本 阿里云DataWorks数据集成是DataX团队在阿里云上的商业化产品，致力于提供复杂网络环境下、丰富的异构数据源之间高速稳定的数据移动能力，以及繁杂业务背景下的数据同步解决方案。目前已经支持云上近3000家客户，单日同步数据超过3万亿条。DataWorks数据集成目前支持离线50+种数据源，可以进行整库迁移、批量上云、增量同步、分库分表等各类同步解决方案。2020年更新实时同步能力，2020年更新实时同步能力，支持10+种数据源的读写任意组合。提供MySQL，Oracle等多种数据源到阿里云MaxCompute，Hologres等大数据引擎的一键全增量同步解决方案。 https://www.aliyun.com/product/bigdata/ide Features DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的Reader插件，以及向目标端写入数据的Writer插件，理论上DataX框架可以支持任意数据源类型的数据同步工作。同时DataX插件体系作为一套生态系统, 每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。 DataX详细介绍 请参考：DataX-Introduction Quick Start Download DataX下载地址 请点击：Quick Start Support Data Channels DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入，目前支持数据如下图，详情请点击：DataX数据源参考指南 类型 数据源 Reader(读) Writer(写) 文档 RDBMS 关系型数据库 MySQL √ √ 读 、写 Oracle √ √ 读 、写 SQLServer √ √ 读 、写 PostgreSQL √ √ 读 、写 DRDS √ √ 读 、写 通用RDBMS(支持所有关系型数据库) √ √ 读 、写 阿里云数仓数据存储 ODPS √ √ 读 、写 ADS √ 写 OSS √ √ 读 、写 OCS √ √ 读 、写 NoSQL数据存储 OTS √ √ 读 、写 Hbase0.94 √ √ 读 、写 Hbase1.1 √ √ 读 、写 Phoenix4.x √ √ 读 、写 Phoenix5.x √ √ 读 、写 MongoDB √ √ 读 、写 Hive √ √ 读 、写 Cassandra √ √ 读 、写 无结构化数据存储 TxtFile √ √ 读 、写 FTP √ √ 读 、写 HDFS √ √ 读 、写 Elasticsearch √ 写 时间序列数据库 OpenTSDB √ 读 TSDB √ √ 读 、写 阿里云DataWorks数据集成 目前DataX的已有能力已经全部融和进阿里云的数据集成，并且比DataX更加高效、安全，同时数据集成具备DataX不具备的其它高级特性和功能。可以理解为数据集成是DataX的全面升级的商业化用版本，为企业可以提供稳定、可靠、安全的数据传输服务。与DataX相比，数据集成主要有以下几大突出特点： ...

DataX DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、SQL Server、Oracle、PostgreSQL、HDFS、Hive、HBase、OTS、ODPS 等各种异构数据源之间高效的数据同步功能。 Features DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的Reader插件，以及向目标端写入数据的Writer插件，理论上DataX框架可以支持任意数据源类型的数据同步工作。同时DataX插件体系作为一套生态系统, 每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。 System Requirements Linux JDK(1.8以上，推荐1.8) Python(推荐Python2.6.X) Apache Maven 3.x (Compile DataX) Quick Start 工具部署 方法一、直接下载DataX工具包：DataX下载地址 下载后解压至本地某个目录，进入bin目录，即可运行同步作业： $ cd {YOUR_DATAX_HOME}/bin $ python datax.py {YOUR_JOB.json} 自检脚本： python {YOUR_DATAX_HOME}/bin/datax.py {YOUR_DATAX_HOME}/job/job.json 方法二、下载DataX源码，自己编译：DataX源码 (1)、下载DataX源码： $ git clone git@github.com:alibaba/DataX.git (2)、通过maven打包： $ cd {DataX_source_code_home} $ mvn -U clean package assembly:assembly -Dmaven.test.skip=true 打包成功，日志显示如下： [INFO] BUILD SUCCESS [INFO] ----------------------------------------------------------------- [INFO] Total time: 08:12 min [INFO] Finished at: 2015-12-13T16:26:48+08:00 [INFO] Final Memory: 133M/960M [INFO] ----------------------------------------------------------------- 打包成功后的DataX包位于 {DataX_source_code_home}/target/datax/datax/ ，结构如下： $ cd {DataX_source_code_home} $ ls ./target/datax/datax/ bin conf job lib log log_perf plugin script tmp 配置示例：从stream读取数据并打印到控制台 第一步、创建作业的配置文件（json格式） 可以通过命令查看配置模板： python datax.py -r {YOUR_READER} -w {YOUR_WRITER} $ cd {YOUR_DATAX_HOME}/bin $ python datax.py -r streamreader -w streamwriter DataX (UNKNOWN_DATAX_VERSION), From Alibaba ! Copyright (C) 2010-2015, Alibaba Group. All Rights Reserved. Please refer to the streamreader document: https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.md Please refer to the streamwriter document: https://github.com/alibaba/DataX/blob/master/streamwriter/doc/streamwriter.md Please save the following configuration as a json file and use python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json to run the job. { "job": { "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [], "sliceRecordCount": "" } }, "writer": { "name": "streamwriter", "parameter": { "encoding": "", "print": true } } } ], "setting": { "speed": { "channel": "" } } } } 根据模板配置json如下： #stream2stream.json { "job": { "content": [ { "reader": { "name": "streamreader", "parameter": { "sliceRecordCount": 10, "column": [ { "type": "long", "value": "10" }, { "type": "string", "value": "hello，你好，世界-DataX" } ] } }, "writer": { "name": "streamwriter", "parameter": { "encoding": "UTF-8", "print": true } } } ], "setting": { "speed": { "channel": 5 } } } } 第二步：启动DataX $ cd {YOUR_DATAX_DIR_BIN} $ python datax.py ./stream2stream.json 同步结束，显示日志如下： ... 2015-12-17 11:20:25.263 [job-0] INFO JobContainer - 任务启动时刻 : 2015-12-17 11:20:15 任务结束时刻 : 2015-12-17 11:20:25 任务总计耗时 : 10s 任务平均流量 : 205B/s 记录写入速度 : 5rec/s 读出记录总数 : 50 读写失败总数 : 0 ...

DataXDataX ADB PG WriterDataX ADS写入CassandraReader 插件文档CassandraWriter 插件文档Readme.mdDataX插件开发宝典DrdsReader 插件文档DataX DRDSWriterREADME.mdDataX ElasticSearchWriterDataX FtpReader 说明DataX FtpWriter 说明DataX GDBReaderDataX GDBWriterHbase094XReader & Hbase11XReader 插件文档Hbase094XWriter & Hbase11XWriter 插件文档Hbase094XReader & Hbase11XReader 插件文档hbase11xsqlreader 插件文档HBase11xsqlwriter插件文档Hbase094XWriter & Hbase11XWriter 插件文档hbase20xsqlreader 插件文档HBase20xsqlwriter插件文档DataX HdfsReader 插件文档DataX HdfsWriter 插件文档阿里云开源离线同步工具DataX3.0介绍KingbaseesReader 插件文档DataX KingbaseesWriterdatax-kudu-plugindatax-kudu-pluginsDatax MongoDBReaderDatax MongoDBWriterMysqlReader 插件文档DataX MysqlWriterDataX OCSWriter 适用memcached客户端写入ocsDataX ODPSReaderDataX ODPS写入OpenTSDBReader 插件文档OracleReader 插件文档DataX OracleWriterDataX OSSReader 说明DataX OSSWriter 说明OTSReader 插件文档TableStore增量数据导出通道：TableStoreStreamReaderOTSWriter 插件文档PostgresqlReader 插件文档DataX PostgresqlWriterRDBMSReader 插件文档RDBMSWriter 插件文档SqlServerReader 插件文档DataX SqlServerWriterDataX TransformerTSDBReader 插件文档TSDBWriter 插件文档DataX TxtFileReader 说明DataX TxtFileWriter 说明DataX

DataX ADB PG Writer 1 快速介绍 AdbpgWriter 插件实现了写入数据到 ABD PG版数据库的功能。在底层实现上，AdbpgWriter 插件会先缓存需要写入的数据，当缓存的 数据量达到 commitSize 时，插件会通过 JDBC 连接远程 ADB PG版 数据库，并执行 COPY 命令将数据写入 ADB PG 数据库。 AdbpgWriter 可以作为数据迁移工具为用户提供服务。 2 实现原理 AdbpgWriter 通过 DataX 框架获取 Reader 生成的协议数据，首先会将数据缓存，当缓存的数据量达到commitSize时，插件根据你配置生成相应的COPY语句，执行 COPY命令将数据写入ADB PG数据库中。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 AdbpgWriter导入的数据 { "job": { "setting": { "speed": { "channel": 32 } }, "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": "adbpgwriter", "parameter": { "username": "username", "password": "password", "host": "host", "port": "1234", "database": "database", "schema": "schema", "table": "table", "preSql": ["delete * from table"], "postSql": ["select * from table"], "column": ["*"] } } } ] } } 3.2 参数说明 name 描述：插件名称 必选：是 默认值：无 username 描述：目的数据库的用户名 必选：是 默认值：无 password 描述：目的数据库的密码 必选：是 默认值：无 host 描述：目的数据库主机名 必选：是 默认值：无 port 描述：目的数据库的端口 必选：是 默认值：无 database 描述：需要写入的表所属的数据库名称 必选：是 默认值：无 schema 描述：需要写入的表所属的schema名称 必选：是 默认值：无 table 描述：需要写入的表名称 必选：是 默认值：无 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 描述：写入数据到目的表后，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，可以使用 @table 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。 必选：否 默认值：否 ...

DataX ADS写入 1 快速介绍 欢迎ADS加入DataX生态圈！ADSWriter插件实现了其他数据源向ADS写入功能，现有DataX所有的数据源均可以无缝接入ADS，实现数据快速导入ADS。 ADS写入预计支持两种实现方式： ADSWriter 支持向ODPS中转落地导入ADS方式，优点在于当数据量较大时(>1KW)，可以以较快速度进行导入，缺点引入了ODPS作为落地中转，因此牵涉三方系统(DataX、ADS、ODPS)鉴权认证。 ADSWriter 同时支持向ADS直接写入的方式，优点在于小批量数据写入能够较快完成(<1KW)，缺点在于大数据导入较慢。 注意： 如果从ODPS导入数据到ADS，请用户提前在源ODPS的Project中授权ADS Build账号具有读取你源表ODPS的权限，同时，ODPS源表创建人和ADS写入属于同一个阿里云账号。 如果从非ODPS导入数据到ADS，请用户提前在目的端ADS空间授权ADS Build账号具备Load data权限。 以上涉及ADS Build账号请联系ADS管理员提供。 2 实现原理 ADS写入预计支持两种实现方式： 2.1 Load模式 DataX 将数据导入ADS为当前导入任务分配的ADS项目表，随后DataX通知ADS完成数据加载。该类数据导入方式实际上是写ADS完成数据同步，由于ADS是分布式存储集群，因此该通道吞吐量较大，可以支持TB级别数据导入。 DataX底层得到明文的 jdbc://host:port/dbname + username + password + table， 以此连接ADS， 执行show grants; 前置检查该用户是否有ADS中目标表的Load Data或者更高的权限。注意，此时ADSWriter使用用户填写的ADS用户名+密码信息完成登录鉴权工作。 检查通过后，通过ADS中目标表的元数据反向生成ODPS DDL，在ODPS中间project中，以ADSWriter的账户建立ODPS表（非分区表，生命周期设为1-2Day), 并调用ODPSWriter把数据源的数据写入该ODPS表中。 注意，这里需要使用中转ODPS的账号AK向中转ODPS写入数据。 写入完成后，以中转ODPS账号连接ADS，发起Load Data From ‘odps://中转project/中转table/’ [overwrite] into adsdb.adstable [partition (xx,xx=xx)]; 这个命令返回一个Job ID需要记录。 注意，此时ADS使用自己的Build账号访问中转ODPS，因此需要中转ODPS对这个Build账号提前开放读取权限。 连接ADS一分钟一次轮询执行 select state from information_schema.job_instances where job_id like ‘$Job ID’，查询状态，注意这个第一个一分钟可能查不到状态记录。 Success或者Fail后返回给用户，然后删除中转ODPS表，任务结束。 上述流程是从其他非ODPS数据源导入ADS流程，对于ODPS导入ADS流程使用如下流程： 2.2 Insert模式 DataX 将数据直连ADS接口，利用ADS暴露的INSERT接口直写到ADS。该类数据导入方式写入吞吐量较小，不适合大批量数据写入。有如下注意点： ADSWriter使用JDBC连接直连ADS，并只使用了JDBC Statement进行数据插入。ADS不支持PreparedStatement，故ADSWriter只能单行多线程进行写入。 ADSWriter支持筛选部分列，列换序等功能，即用户可以填写列。 考虑到ADS负载问题，建议ADSWriter Insert模式建议用户使用TPS限流，最高在1W TPS。 ADSWriter在所有Task完成写入任务后，Job Post单例执行flush工作，保证数据在ADS整体更新。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到ADS，使用Load模式进行导入的数据。 { "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ { "value": "DataX", "type": "string" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 100000 } }, "writer": { "name": "adswriter", "parameter": { "odps": { "accessId": "xxx", "accessKey": "xxx", "account": "xxx@aliyun.com", "odpsServer": "xxx", "tunnelServer": "xxx", "accountType": "aliyun", "project": "transfer_project" }, "writeMode": "load", "url": "127.0.0.1:3306", "schema": "schema", "table": "table", "username": "username", "password": "password", "partition": "", "lifeCycle": 2, "overWrite": true, } } } ] } } 这里使用一份从内存产生到ADS，使用Insert模式进行导入的数据。 { "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "streamreader", "parameter": { "column": [ { "value": "DataX", "type": "string" }, { "value": "test", "type": "bytes" } ], "sliceRecordCount": 100000 } }, "writer": { "name": "adswriter", "parameter": { "writeMode": "insert", "url": "127.0.0.1:3306", "schema": "schema", "table": "table", "column": ["*"], "username": "username", "password": "password", "partition": "id,ds=2015" } } } ] } } 3.2 参数说明 （用户配置规格） url ...

DataX DRDSWriter 1 快速介绍 DRDSWriter 插件实现了写入数据到 DRDS 的目的表的功能。在底层实现上， DRDSWriter 通过 JDBC 连接远程 DRDS 数据库的 Proxy，并执行相应的 replace into … 的 sql 语句将数据写入 DRDS，特别注意执行的 Sql 语句是 replace into，为了避免数据重复写入，需要你的表具备主键或者唯一性索引(Unique Key)。 DRDSWriter 面向ETL开发工程师，他们使用 DRDSWriter 从数仓导入数据到 DRDS。同时 DRDSWriter 亦可以作为数据迁移工具为DBA等用户提供服务。 2 实现原理 DRDSWriter 通过 DataX 框架获取 Reader 生成的协议数据，通过 replace into...(没有遇到主键/唯一性索引冲突时，与 insert into 行为一致，冲突时会用新行替换原有行所有字段) 的语句写入数据到 DRDS。DRDSWriter 累积一定数据，提交给 DRDS 的 Proxy，该 Proxy 内部决定数据是写入一张还是多张表以及多张表写入时如何路由数据。 注意：整个任务至少需要具备 replace into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 3 功能说明 3.1 配置样例 这里使用一份从内存产生到 DRDS 导入的数据。 { "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": "drdswriter", "parameter": { "writeMode": "insert", "username": "root", "password": "root", "column": [ "id", "name" ], "preSql": [ "delete from test" ], "connection": [ { "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/datax?useUnicode=true&characterEncoding=gbk", "table": [ "test" ] } ] } } } ] } } 3.2 参数说明 jdbcUrl ...

DataX ElasticSearchWriter 1 快速介绍 数据导入elasticsearch的插件 2 实现原理 使用elasticsearch的rest api接口， 批量把从reader读入的数据写入elasticsearch 3 功能说明 3.1 配置样例 job.json { "job": { "setting": { "speed": { "channel": 1 } }, "content": [ { "reader": { ... }, "writer": { "name": "elasticsearchwriter", "parameter": { "endpoint": "http://xxx:9999", "accessId": "xxxx", "accessKey": "xxxx", "index": "test-1", "type": "default", "cleanup": true, "settings": {"index" :{"number_of_shards": 1, "number_of_replicas": 0}}, "discovery": false, "batchSize": 1000, "splitter": ",", "column": [ {"name": "pk", "type": "id"}, { "name": "col_ip","type": "ip" }, { "name": "col_double","type": "double" }, { "name": "col_long","type": "long" }, { "name": "col_integer","type": "integer" }, { "name": "col_keyword", "type": "keyword" }, { "name": "col_text", "type": "text", "analyzer": "ik_max_word"}, { "name": "col_geo_point", "type": "geo_point" }, { "name": "col_date", "type": "date", "format": "yyyy-MM-dd HH:mm:ss"}, { "name": "col_nested1", "type": "nested" }, { "name": "col_nested2", "type": "nested" }, { "name": "col_object1", "type": "object" }, { "name": "col_object2", "type": "object" }, { "name": "col_integer_array", "type":"integer", "array":true}, { "name": "col_geo_shape", "type":"geo_shape", "tree": "quadtree", "precision": "10m"} ] } } } ] } } 3.2 参数说明 endpoint 描述：ElasticSearch的连接地址 必选：是 默认值：无 accessId 描述：http auth中的user 必选：否 默认值：空 accessKey 描述：http auth中的password 必选：否 默认值：空 index 描述：elasticsearch中的index名 必选：是 默认值：无 type 描述：elasticsearch中index的type名 必选：否 默认值：index名 cleanup 描述：是否删除原表 必选：否 默认值：false batchSize 描述：每次批量数据的条数 必选：否 默认值：1000 trySize 描述：失败后重试的次数 必选：否 默认值：30 timeout 描述：客户端超时时间 必选：否 默认值：600000 discovery 描述：启用节点发现将(轮询)并定期更新客户机中的服务器列表。 必选：否 默认值：false compression 描述：http请求，开启压缩 必选：否 默认值：true multiThread 描述：http请求，是否有多线程 必选：否 默认值：true ignoreWriteError 描述：忽略写入错误，不重试，继续写入 必选：否 默认值：false ignoreParseError 描述：忽略解析数据格式错误，继续写入 必选：否 默认值：true alias 描述：数据导入完成后写入别名 必选：否 默认值：无 aliasMode 描述：数据导入完成后增加别名的模式，append(增加模式), exclusive(只留这一个) 必选：否 默认值：append settings 描述：创建index时候的settings, 与elasticsearch官方相同 必选：否 默认值：无 splitter 描述：如果插入数据是array，就使用指定分隔符 必选：否 默认值：-,- column 描述：elasticsearch所支持的字段类型，样例中包含了全部 必选：是 dynamic 描述: 不使用datax的mappings，使用es自己的自动mappings 必选: 否 默认值: false ...

DataX FtpReader 说明 1 快速介绍 FtpReader提供了读取远程FTP文件系统数据存储的能力。在底层实现上，FtpReader获取远程FTP文件数据，并转换为DataX传输协议传递给Writer。 本地文件内容存放的是一张逻辑意义上的二维表，例如CSV格式的文本信息。 2 功能与限制 FtpReader实现了从远程FTP文件读取数据并转为DataX协议的功能，远程FTP文件本身是无结构化数据存储，对于DataX而言，FtpReader实现上类比TxtFileReader，有诸多相似之处。目前FtpReader支持功能如下： 支持且仅支持读取TXT的文件，且要求TXT中shema为一张二维表。 支持类CSV格式文件，自定义分隔符。 支持多种类型数据读取(使用String表示)，支持列裁剪，支持列常量 支持递归读取、支持文件名过滤。 支持文本压缩，现有压缩格式为zip、gzip、bzip2。 多个File可以支持并发读取。 我们暂时不能做到： 单个File支持多线程并发读取，这里涉及到单个File内部切分算法。二期考虑支持。 单个File在压缩情况下，从技术上无法支持多线程并发读取。 3 功能说明 3.1 配置样例 { "setting": {}, "job": { "setting": { "speed": { "channel": 2 } }, "content": [ { "reader": { "name": "ftpreader", "parameter": { "protocol": "sftp", "host": "127.0.0.1", "port": 22, "username": "xx", "password": "xxx", "path": [ "/home/hanfa.shf/ftpReaderTest/data" ], "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" } ], "encoding": "UTF-8", "fieldDelimiter": "," } }, "writer": { "name": "ftpWriter", "parameter": { "path": "/home/hanfa.shf/ftpReaderTest/result", "fileName": "shihf", "writeMode": "truncate", "format": "yyyy-MM-dd" } } } ] } } 3.2 参数说明 protocol 描述：ftp服务器协议，目前支持传输协议有ftp和sftp。 必选：是 默认值：无 host 描述：ftp服务器地址。 必选：是 默认值：无 port 描述：ftp服务器端口。 必选：否 默认值：若传输协议是sftp协议，默认值是22；若传输协议是标准ftp协议，默认值是21 timeout 描述：连接ftp服务器连接超时时间，单位毫秒。 必选：否 默认值：60000（1分钟） connectPattern 描述：连接模式（主动模式或者被动模式）。该参数只在传输协议是标准ftp协议时使用，值只能为：PORT (主动)，PASV（被动）。两种模式主要的不同是数据连接建立的不同。对于Port模式，是客户端在本地打开一个端口等服务器去连接建立数据连接，而Pasv模式就是服务器打开一个端口等待客户端去建立一个数据连接。 必选：否 默认值：PASV username 描述：ftp服务器访问用户名。 必选：是 默认值：无 password 描述：ftp服务器访问密码。 必选：是 默认值：无 path 描述：远程FTP文件系统的路径信息，注意这里可以支持填写多个路径。 当指定单个远程FTP文件，FtpReader暂时只能使用单线程进行数据抽取。二期考虑在非压缩文件情况下针对单个File可以进行多线程并发读取。 当指定多个远程FTP文件，FtpReader支持使用多线程进行数据抽取。线程并发数通过通道数指定。 当指定通配符，FtpReader尝试遍历出多个文件信息。例如: 指定/*代表读取/目录下所有的文件，指定/bazhen/*代表读取bazhen目录下游所有的文件。FtpReader目前只支持*作为文件通配符。 特别需要注意的是，DataX会将一个作业下同步的所有Text File视作同一张数据表。用户必须自己保证所有的File能够适配同一套schema信息。读取文件用户必须保证为类CSV格式，并且提供给DataX权限可读。 特别需要注意的是，如果Path指定的路径下没有符合匹配的文件抽取，DataX将报错。 必选：是 默认值：无 column 描述：读取字段列表，type指定源数据的类型，index指定当前列来自于文本第几列(以0开始)，value指定当前类型为常量，不从源头文件读取数据，而是根据value值自动生成对应的列。 默认情况下，用户可以全部按照String类型读取数据，配置如下： "column": ["*"] 用户可以指定Column字段信息，配置如下： { "type": "long", "index": 0 //从远程FTP文件文本第一列获取int字段 }, { "type": "string", "value": "alibaba" //从FtpReader内部生成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 maxTraversalLevel 描述：允许遍历文件夹的最大层数。 必选：否 默认值：100 csvReaderConfig 描述：读取CSV类型文件参数配置，Map类型。读取CSV类型文件使用的CsvReader进行读取，会有很多配置，不配置则使用默认值。 必选：否 默认值：无 ...