0
0

flume对kafka中数据的导入导出、datax对mysql数据库数据的抽取

要求：

1、flume自定义拦截器

抽取trans_info.json的数据到kafka上，对其中的tr_flag=0的数据进行过滤抛弃，只保留正常的状态数据

在pom.xml中放入依赖包： 

  1. <dependency>
  2.             <groupId>org.apache.kafka</groupId>
  3.             <artifactId>kafka-clients</artifactId>
  4.             <version>3.0.0</version>
  5.         </dependency>
  7.         <dependency>
  8.             <groupId>org.slf4j</groupId>
  9.             <artifactId>slf4j-log4j12</artifactId>
  10.             <version>1.7.25</version>
  11.         </dependency>
  13.         <dependency>
  14.             <groupId>org.apache.hadoop</groupId>
  15.             <artifactId>hadoop-common</artifactId>
  16.             <version>2.6.5</version>
  17.         </dependency>
  18.         <dependency>
  19.             <groupId>org.apache.hadoop</groupId>
  20.             <artifactId>hadoop-hdfs</artifactId>
  21.             <version>2.6.5</version>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.apache.hadoop</groupId>
  25.             <artifactId>hadoop-client</artifactId>
  26.             <version>2.6.5</version>
  27.         </dependency>
  29.         <dependency>
  30.             <groupId>org.apache.flume</groupId>
  31.             <artifactId>flume-ng-core</artifactId>
  32.             <version>1.9.0</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.apache.flume</groupId>
  36.             <artifactId>flume-ng-sdk</artifactId>
  37.             <version>1.9.0</version>
  38.         </dependency>
  40.         <dependency>
  41.             <groupId>com.alibaba</groupId>
  42.             <artifactId>fastjson</artifactId>
  43.             <version>1.2.48</version>
  44.         </dependency>

使用java代码，自定义拦截器： 

  1. package com.bigdata.yuekao04;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONArray;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flume.Context;
  7. import org.apache.flume.Event;
  8. import org.apache.flume.interceptor.Interceptor;
  9. import org.codehaus.jackson.JsonNode;
  10. import org.codehaus.jackson.map.ObjectMapper;
  12. import java.io.IOException;
  13. import java.util.ArrayList;
  14. import java.util.HashMap;
  15. import java.util.List;
  17. public class DemoInterceptor implements Interceptor {
  18.     @Override
  19.     public void initialize() {
  21.     }
  23.     @Override
  24.     public Event intercept(Event event) {
  25.         try {
  26.             // 获取事件体中的数据（假设数据是JSON格式存储在事件体中）
  27.             String data = new String(event.getBody());
  29.             // 使用Jackson将JSON字符串解析为JsonNode对象
  30.             ObjectMapper objectMapper = new ObjectMapper();
  31.             JsonNode jsonNode = objectMapper.readTree(data);
  33.             // 获取tr_flag的值
  34.             int trFlag = jsonNode.get("tr_flag").asInt();
  36.             // 如果tr_flag不等于0，保留该事件
  37.             if (trFlag!= 0) {
  38.                 return event;
  39.             }
  40.         } catch (IOException e) {
  41.             e.printStackTrace();
  42.         }
  44.         // 如果tr_flag等于0，返回null，表示过滤掉该事件
  45.         return null;
  46.         }
  48.     @Override
  49.     public List<Event> intercept(List<Event> list) {
  50.         return null;
  51.     }
  53.     @Override
  54.     public void close() {
  56.     }
  58.     public static class BuilderEvent implements Builder{
  60.         @Override
  61.         public Interceptor build() {
  62.             return new DemoInterceptor();
  63.         }
  65.         @Override
  66.         public void configure(Context context) {
  68.         }
  69.     }
  70. }

打包java代码，放入/flume/lib下面

2、创建topic为yuekao的主题，并使用flume将数据抽取到该主题的kafka中

编写conf文件（yuekao04.conf），将数据抽取到kafka新创建的主题中：

  1. # 定义Flume agent名称
  2. a1.sources = r1
  3. a1.sinks = k1
  4. a1.channels = c1
  6. # 配置source
  7. a1.sources.r1.type = TAILDIR
  8. #以空格分隔的文件组列表。每个文件组表示要跟踪的一组文件
  9. a1.sources.s1.filegroups = f1
  10. #文件组的绝对路径
  11. a1.sources.s1.filegroups.f1=/home/trans_info1.json
  12. a1.sources.r1.interceptors = i1
  13. a1.sources.r1.interceptors.i1.type = com.bigdata.DemoInterceptor$Builder
  15. # 配置channel
  16. a1.channels.c1.type = file
  18. # 配置sink
  19. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  20. a1.sinks.k1.kafka.topic = yuekao04
  21. a1.sinks.k1.kafka.bootstrap.servers = bigdata01:9092
  22. a1.sinks.k1.channel = c1
3、将kafka中的数据放入到hdfs上，目录为：/yuekao/ods/zhuanzhang

编写conf文件，然后执行该文件，将kafka中的数据放入hdfs中： 

  1. a1.sources = r1
  2. a1.channels = c1
  3. a1.sinks=k1
  5. a1.sources.r1.channels = c1
  6. a1.sinks.k1.channel = c1
  8. a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
  9. a1.sources.r1.kafka.bootstrap.servers = bigdata01:9092,bigdata02:9092,bigdata03:9092
  10. a1.sources.r1.kafka.topics = yuekao04
  11. a1.sources.r1.kafka.consumer.group.id =yuekao
  12. a1.sources.r1.batchSize = 100
  13. a1.sources.r1.batchDurationMillis = 2000
  15. a1.channels.c1.type = memory
  16. a1.channels.c1.capacity = 1000
  17. a1.channels.c1.transactionCapacity = 100
  19. a1.sinks.k1.type = hdfs
  20. a1.sinks.k1.channel = c1
  21. a1.sinks.k1.hdfs.path = /yuekao/ods/zhuanzhang/%y-%m-%d
  22. a1.sinks.k1.hdfs.filePrefix = events-
  23. a1.sinks.k1.hdfs.round = true
  24. a1.sinks.k1.hdfs.roundValue = 10
  25. a1.sinks.k1.hdfs.roundUnit = minute
  26. a1.sinks.k1.hdfs.fileType = DataStream

结果展示：

4、 通过datax，对MySQL数据库中的表进行抽取，落入hdfs指定的目录中: /yuekao/ods/user_info

先在mysql中建表，然后将user_info.sql表中数据插入：

  1. CREATE TABLE `user_info` (
  2. `name` VARCHAR (255) ,
  3. phone_num VARCHAR (255) ,
  4. email VARCHAR (255) ,
  5. addr_info VARCHAR (255) ,
  6. gender VARCHAR (255) ,
  7. idno VARCHAR (255) ,
  8. create_time VARCHAR (255) ,
  9. user_id int
  10. );

编写json文件（demo.json），然后执行，将数据库中的数据放入hdfs中：

  1. {
  2.     "job": {
  3.         "setting": {
  4.             "speed": {
  5.                  "channel": 3
  6.             },
  7.             "errorLimit": {
  8.                 "record": 0,
  9.                 "percentage": 0.02
  10.             }
  11.         },
  12.         "content": [
  13.             {
  14.                 "reader": {
  15.                     "name": "mysqlreader",
  16.                     "parameter": {
  17.                         "writeMode": "insert",
  18.                         "username": "root",
  19.                         "password": "123456",
  20.                         "column": [
  21.                             "name",
  22.                             "phone_num",
  23.                             "email",
  24.                             "addr_info",
  25.                             "gender",
  26.                             "idno",
  27.                             "create_time",
  28.                             "user_id"
  29.                         ],
  30.                         "splitPk": "user_id",
  31.                         "connection": [
  32.                             {
  33.                                 "table": [
  34.                                     "user_info"
  35.                                 ],
  36.                                 "jdbcUrl": [
  37.      "jdbc:mysql://bigdata01:3306/yuekao"
  38.                                 ]
  39.                             }
  40.                         ]
  41.                     }
  42.                 },
  43.                "writer": {
  44.                     "name": "hdfswriter",
  45.                     "parameter": {
  46.                         "defaultFS": "hdfs://bigdata01:9820",
  47.                         "fileType": "text",
  48.                         "path": "/yuekao/ods/user_info",
  49.                         "fileName": "user_info.txt",
  50.                         "column": [
  51.                             {
  52.                                 "name": "name",
  53.                                 "type": "string"
  54.                             },
  55.                             {
  56.                                 "name": "phone_num",
  57.                                 "type": "string"
  58.                             },
  59.                             {
  60.                                 "name": "email",
  61.                                 "type": "string"
  62.                             },
  63.                             {
  64.                                 "name": "addr_info",
  65.                                 "type": "string"
  66.                             },
  67.                             {
  68.                                 "name": "gender",
  69.                                 "type": "string"
  70.                             },
  71.                             {
  72.                                 "name": "idno",
  73.                                 "type": "string"
  74.                             },
  75.                             {
  76.                                 "name": "create_time",
  77.                                 "type": "string"
  78.                             },
  79.                             {
  80.                                 "name": "user_id",
  81.                                 "type": "int"
  82.                             }
  83.                         ],
  84.                         "writeMode": "append",
  85.                         "fieldDelimiter": ","
  86.                     }
  87.                 }
  88.             }
  89.         ]
  90.     }
  91. }

执行json文件： 

  1. datax.py demo.json

结果展示：

数据放不进来，有需要的小伙伴可以私我！！！

标签： 数据库 flume kafka
本文转载自: https://blog.csdn.net/weixin_64860388/article/details/143835597
版权归原作者 jlting195 所有， 如有侵权，请联系我们删除。
登录后发布评论

“flume对kafka中数据的导入导出、datax对mysql数据库数据的抽取”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。