DataX读取Hive数据的话,其本身只提供了hdfsreader,因为hive一来不是个数据库,它只是hdfs数据的结构化管理工具,所以你可以直接用hdfsreader,只能列名的方式抽hive数据,一般用到的抽取场景如下几类。
第一种:全字段数据,源数据hive,目的库关系型数据库,比如mysql。全表时hdfsreader的column可以简写为*
{"job": {"content": [{"reader": {"name": "hdfsreader", "parameter": {"column": ["*"], "defaultFS": "hdfs://hdp1:9000","encoding": "UTF-8","fieldDelimiter": ",","fileType": "text","path": "/hiveData/test"}}, "writer": {"name": "mysqlwriter", "parameter": {"column": ["id","name","sex"], "connection": [{"jdbcUrl": "jdbc:mysql://192.168.0.103:3306/shop?useUnicode=true&characterEncoding=utf8", "table": ["test"]}], "password": "123456", "username": "root", "writeMode": "insert"}}}], "setting": {"speed": {"channel": "1"}}}
}
第二种,部分字段,源数据hive,目的库关系型数据库,比如mysql。
{"job": {"content": [{"reader": {"name": "hdfsreader", "parameter": {"column": [{"index":1,"name": "name", "type": "string"},{"index":2,"name": "sex", "type": "string"},], "defaultFS": "hdfs://hdp1:9000","encoding": "UTF-8","fieldDelimiter": ",","fileType": "text","path": "/hiveData/test"}}, "writer": {"name": "mysqlwriter", "parameter": {"column": ["name","sex"], "connection": [{"jdbcUrl": "jdbc:mysql://192.168.0.103:3306/shop?useUnicode=true&characterEncoding=utf8", "table": ["test"]}], "password": "123456", "username": "root", "writeMode": "insert"}}}], "setting": {"speed": {"channel": "1"}}}
}
如果你要对,目的端在数据落库之前做一些预处理,可以在writer的Json部分写如下配置,比如要删掉目的表中的一些数据
"preSql": ["delete from paper_avgtimeandscore where s='1' "
]
从hive数据里抽,一般就上面这两种情况,注意原生情况下hdfsreader是没有提供数据过滤能力,就是where,因为抽取的时候一般都是按分区抽,或者干脆就是全量,对于where的需求在hive里面就已经解决了,一般是做一个dwd报表,说白了数据从hive出来的时候就没有where的业务必要,所以hdfsreader就不含有这种能力,但是
市场上存在第三方的reader插件可以完成这种能力,如果有需要自己找一找就行。
第三种:从其他数据端抽取数据落到hive中,比如从mysql抽,最后落库到hive
{"job": {"setting": {"speed": {"channel": 1}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","connection": [{"jdbcUrl": ["jdbc:mysql://192.168.0.103:3306/shop?useUnicode=true&characterEncoding=utf8"],"querySql": ["SELECT id, name, sex FROM your_table_name"]}]}},"writer": {"name": "hdfswriter","parameter": {"defaultFS": "hdfs://hdp1:9000","fileType": "text","path": "/hiveData/test","fileName": "part-0101","column": [{"name": "id", "type": "string"},{"name": "name", "type": "string"},{"name": "sex", "type": "string"}],"fieldDelimiter": ",","writeFormat": "text","writeMode": "append"}}}]}
}
无论是你导入还导出一定要注意的点:
只要涉及到两端的列明定义一定要一一对应,比方说hdfsreader的column中,你可以不定义name属性,但必须定义index,index的值是hdfs文件中列的下标,并且每一个column中的Json对象,要和输出端,如在本例中是mysqlwriter的column部分一一对应,不能错列,就是说你reader端第一个column定义的是hdfs文件中下标为2的列,那么下面输出端的column中第一个也必须是hdfs文件中下标为2这一列数据你希望对应的列,反过来也是一样的,DataX不会给你自动识别位置的。
在使用DataX的时候,对于高可用的Hadoop集群,要注意一点,我上面写的例子都是直接指定的namenode,如果你要把抽取程序运行到高可用的集群上的话,就要在hdfswriter或hdfsreader的parameter
中加如下配置,既高可用namenode节点的配置信息,当然配置改成你自己的
"hadoopConfig":{"dfs.nameservices": "mycluster","dfs.ha.namenodes.mycluster": "nn1,nn2","dfs.namenode.rpc-address.mycluster.nn1": "hadoop101:8020","dfs.namenode.rpc-address.mycluster.nn2": "hadoop102:8020","dfs.client.failover.proxy.provider.mycluster": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
}
然后defaultFS也就写高可用逻辑组名就行
"defaultFS": "hdfs://mycluster",
之所以要这么干,是因为DataX不去识别你的本地Hadoop配置,或者是HOME,它本身就允许你不在Hadoop集群节点上跑数据。