当前位置：首页 > article >正文

DataX用hdfsreader导入或导出hive数据

article 2024/11/15 4:34:56

DataX读取Hive数据的话，其本身只提供了hdfsreader，因为hive一来不是个数据库，它只是hdfs数据的结构化管理工具，所以datax默认没有自带hive的reader，因此默认只能直接用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部分写如下配置，比如要删掉目的mysql表中的一些数据

"preSql": [
     "delete from paper_avgtimeandscore where s='1' "
]

从hive数据里抽，一般就上面这两种情况，注意原生情况下hdfsreader是没有提供数据过滤能力，就是where，因为抽取的时候一般都是按分区抽，或者干脆就是全量，对于where的需求在hive里面就已经解决了，一般是做一个dwd报表，说白了数据从hive出来的时候就没有where的业务必要，所以hdfsreader就不含有这种能力，但是市场上存在第三方的reader插件可以完成这种能力，如果有需要自己找一找就行，不过多数是自己开发一个hivereader。

第三种：从其他数据端抽取数据落到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不会给你自动识别位置的。最重要的是如果你用的是hdfsreader，要明白这个组件本质上是用来处理hdfs的，因此它没办法在其他数据库导入数据到hive时，完成部分字段导入，这一点就要去自定义hivereader了，你直接用hdfswriter写的话会发现数据任然是顺序依次的系列化，和列明对不上的。而hive数据导出到其他数据库的时候可以部分字段导，因为本质上还是用的对应数据库的jdbc。

在使用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"
}