DataX的如何使用hdfsreader/writer

2024-09-07 16:20
文章标签 使用 datax writer hdfsreader

本文主要是介绍DataX的如何使用hdfsreader/writer,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

说明:DataX的hdfs读取或写入一般用的比较少,国内用datax通常都是用它完成数据仓库之间的数据迁移,很少以文件的形式直接迁移,对于hdfs来讲,datax提供了hdfsreader和hdfswriter,本篇以文件的方式导入或导出hive数据为例,展示datax的hdfsreader/writer怎么用,因为整体的技术大环境下使得datax提供的hdfsreader/writer也是以结构化数据的方式传递文件,但是要提前说明的是,虽然hive不是数据库,它只是hdfs数据的结构化管理工具,但是hive支持jdbc的形式使用,所以datax默认没有自带所谓的hivereader/writer,如果你的技术环境很干净的场景下,例如你使用的就是原生的或者hdp这种定制化不高,一般用在基础架构上的技术环境的话,你可以直接采用jdbcreader/writer去迁移hive数据。本篇只是为了展示hdfsreader/writer如何写任务Json配置,所以借用hive为使用例子,因为它底层就是hdfs的文件。同样的正是因为datax本身并没有自带所谓的hivereader/writer,所以如大家在工作中使用的中台之类的产品,你看到的日志中输出的hivereader之类的配置,那都是中台自己的RD二次开发的,同时datax本身也支持用户二次开发reader/writer。言归正传,用hive为例hdfsreader/writer使用方式如下。

第一种:全字段数据,源数据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就不含有这种能力,除非向开头说的那样市场上存在的第三方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不会给你自动识别位置的,因为人家本身就是为了文件传递而存在,开发用意上就没考虑hive。因此它没办法在其他数据库导入数据到hive时,完成部分字段导入,要实现这一点就要去自定义hivereader或者用jdbc了,你直接用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"
}

然后上面json文件里defaultFS也就写高可用逻辑组名就行

"defaultFS": "hdfs://mycluster",

之所以要这么干,是因为DataX不去识别你的本地Hadoop配置,或者是HOME,它本身就允许你不在Hadoop集群节点上跑数据。

这篇关于DataX的如何使用hdfsreader/writer的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1145551

相关文章

sky-take-out项目中Redis的使用示例详解

《sky-take-out项目中Redis的使用示例详解》SpringCache是Spring的缓存抽象层,通过注解简化缓存管理,支持Redis等提供者,适用于方法结果缓存、更新和删除操作,但无法实现... 目录Spring Cache主要特性核心注解1.@Cacheable2.@CachePut3.@Ca

C#下Newtonsoft.Json的具体使用

《C#下Newtonsoft.Json的具体使用》Newtonsoft.Json是一个非常流行的C#JSON序列化和反序列化库,它可以方便地将C#对象转换为JSON格式,或者将JSON数据解析为C#对... 目录安装 Newtonsoft.json基本用法1. 序列化 C# 对象为 JSON2. 反序列化

RabbitMQ 延时队列插件安装与使用示例详解(基于 Delayed Message Plugin)

《RabbitMQ延时队列插件安装与使用示例详解(基于DelayedMessagePlugin)》本文详解RabbitMQ通过安装rabbitmq_delayed_message_exchan... 目录 一、什么是 RabbitMQ 延时队列? 二、安装前准备✅ RabbitMQ 环境要求 三、安装延时队

Python ORM神器之SQLAlchemy基本使用完全指南

《PythonORM神器之SQLAlchemy基本使用完全指南》SQLAlchemy是Python主流ORM框架,通过对象化方式简化数据库操作,支持多数据库,提供引擎、会话、模型等核心组件,实现事务... 目录一、什么是SQLAlchemy?二、安装SQLAlchemy三、核心概念1. Engine(引擎)

Java Stream 并行流简介、使用与注意事项小结

《JavaStream并行流简介、使用与注意事项小结》Java8并行流基于StreamAPI,利用多核CPU提升计算密集型任务效率,但需注意线程安全、顺序不确定及线程池管理,可通过自定义线程池与C... 目录1. 并行流简介​特点:​2. 并行流的简单使用​示例:并行流的基本使用​3. 配合自定义线程池​示

GO语言中函数命名返回值的使用

《GO语言中函数命名返回值的使用》在Go语言中,函数可以为其返回值指定名称,这被称为命名返回值或命名返回参数,这种特性可以使代码更清晰,特别是在返回多个值时,感兴趣的可以了解一下... 目录基本语法函数命名返回特点代码示例命名特点基本语法func functionName(parameters) (nam

使用shardingsphere实现mysql数据库分片方式

《使用shardingsphere实现mysql数据库分片方式》本文介绍如何使用ShardingSphere-JDBC在SpringBoot中实现MySQL水平分库,涵盖分片策略、路由算法及零侵入配置... 目录一、ShardingSphere 简介1.1 对比1.2 核心概念1.3 Sharding-Sp

Java 正则表达式的使用实战案例

《Java正则表达式的使用实战案例》本文详细介绍了Java正则表达式的使用方法,涵盖语法细节、核心类方法、高级特性及实战案例,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要... 目录一、正则表达式语法详解1. 基础字符匹配2. 字符类([]定义)3. 量词(控制匹配次数)4. 边

Python Counter 函数使用案例

《PythonCounter函数使用案例》Counter是collections模块中的一个类,专门用于对可迭代对象中的元素进行计数,接下来通过本文给大家介绍PythonCounter函数使用案例... 目录一、Counter函数概述二、基本使用案例(一)列表元素计数(二)字符串字符计数(三)元组计数三、C

使用Spring Cache本地缓存示例代码

《使用SpringCache本地缓存示例代码》缓存是提高应用程序性能的重要手段,通过将频繁访问的数据存储在内存中,可以减少数据库访问次数,从而加速数据读取,:本文主要介绍使用SpringCac... 目录一、Spring Cache简介核心特点:二、基础配置1. 添加依赖2. 启用缓存3. 缓存配置方案方案