采集方式汇集(sqoop、spark、flume、logstash、filebeat)

Bertha 。 2023-02-15 03:33 33阅读 0赞

# 前文： #

数据仓库中ods层一般使用外部表，一般默认采用 \\001 作为分隔符，但textfile格式在遇到文本数据就会有分隔符及换行符问题，所以采用parquet作为存储格式，但也会引进数据类型转换的问题。

对于日志数据，kafka中经常存放不同来源的日志数据，可通过Flume的正则匹配将数据发送到不同的hdfs文件夹。

对于部分mysql百万级进行模糊查询将会产生慢查询，一般我们用可以存放在es中。

# 一、Mysql导数据导Hive #

## 1.1 建表 ##

create external table if not exists ods.ods_stu( 
    `id` int comment '主键Id', 
    `name` string comment '名称',
    `addtime` string commint '添加时间'
    )comment '姓名表'
    partitioned by (dt string)
    stored as parquet
    location '/user/hive/warehouse/ods.db/ods_stu';

备注：parquet存储格式查询效率高

## 1.2 使用sqoop导数据 ##

sqoop import \
    --connect jdbc:mysql://localhost:3306/ods_stu\
    --username root\
    --password 123456\
    --null-string 'NULL'  \
    --null-non-string 'NULL' \
    --query '
    select * 
    from stu
    where $CONDITIONS' \
    --target-dir /user/hive/warehouse/ods.db/ods_stu/dt=2020-06-04 \
    --map-column-java addtime=String \
    --map-column-hive addtime=String 
    --append \
    --split-by id \
    --num-mappers 2 \
    --as-parquetfile \

存在问题：sqoop直接从mysql导入parquet格式的hive表有一个数据类型异常

Failed with exception java.io.IOException:org.apache.hadoop.hive.ql.metadata.HiveException: java.lang.ClassCastException: org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.hive.serde2.io.TimestampWritabl

解决方式：采用 --map-column-java 和 --map-column-hive 把mysql中时间类型之间转换为string类型存储hive中

load data inpath '/user/hive/warehouse/ods.db/ods_stu/dt=2020-06-04' into table ods.ods_minisns_systemtag partition (dt = '2020-06-04');

## 1.3 使用spark导数据 ##

/**
         * 通过拉取mysql数据，创建临时表写入hive中
         */
        val df: DataFrame = spark.read.format("jdbc")
          .options(mysqlParam)
          //增量：单线程
          .option("query", query_mysql)
          .load()
        df.createOrReplaceTempView(tmp_tablename)
        spark.sql(insert_hive);

备注：spark可以直接写入parquet格式的hive表，并且可以直接数据清洗。但spark连接查询mysql会有一个散列id排序的问题，并且使用query自定义sql查询的话只有1个线程拉数据。而sqoop可以完美的解决这两个问题，使用方式简单。

## 1.4sqoop与spark导数据的比较 ##

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>sqoop</td> 
   <td>spark</td> 
  </tr> 
  <tr> 
   <td>textfile格式外部表</td> 
   <td>不支持复杂分隔符，需要编译</td> 
   <td>无需关心分隔符</td> 
  </tr> 
  <tr> 
   <td>paquet格式</td> 
   <td>需处理数据类型转换</td> 
   <td>无需关心数据类型问题</td> 
  </tr> 
  <tr> 
   <td>散列id同步</td> 
   <td>存在倾斜问题</td> 
   <td> <p>存在倾斜问题</p> </td> 
  </tr> 
  <tr> 
   <td>并行度</td> 
   <td>完美处理</td> 
   <td>query自定义查询只有单线程查询mysql</td> 
  </tr> 
  <tr> 
   <td>直接清洗</td> 
   <td>仅在mysql查询时清洗，性能较差</td> 
   <td>拉取数据后直接清洗</td> 
  </tr> 
  <tr> 
   <td>使用难度</td> 
   <td>简单配置</td> 
   <td>写个工具类传参也可，但较复杂</td> 
  </tr> 
 </tbody> 
</table>

# 二、Flume采集Kafka数据 #

## 2.1kafka数据到hive外部表 ##

a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    #————————————————————  sources  ————————————————————#
    
    a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
    a1.sources.r1.batchSize = 5000
    a1.sources.r1.batchDurationMillis = 2000
    a1.sources.r1.kafka.bootstrap.servers = bigdata-test:9092
    a1.sources.r1.kafka.topics = test_0603
    a1.sources.r1.kafka.consumer.group.id = custom.g.id
    a1.sources.r1.kafka.consumer.auto.offset.reset=earliest
    
    
    
    #————————————————————  interceptors  ————————————————————#
    
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = regex_extractor
    
    # 正则获取捕获组并命名，用于sink文件夹路径设置
    
    a1.sources.r1.interceptors.i1.regex = (\\d{4}\\-\\d{2}\\-\\d{2})
    a1.sources.r1.interceptors.i1.serializers = s1
    a1.sources.r1.interceptors.i1.serializers.s1.name = eventTime
    
    
    
    #————————————————————  channels  ————————————————————#
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 10000
    a1.channels.c1.transactionCapacity = 10000
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    
    
    #————————————————————  sinks  ————————————————————#
    
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = /tmp/flume_test_0603/%{eventTime}
    
    # 使用当地时间(而不是从事件的时间戳头)而取代转义序列。
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    
    # DataStream不会压缩输出文件
    a1.sinks.k1.hdfs.fileType = DataStream	
    
    # 默认为1024，触发滚动的文件大小，以字节为单位（0：从不基于文件大小滚动）
    a1.sinks.k1.hdfs.rollSize = 104857600  
    
    # 默认为30，滚动当前文件之前要等待的秒数（0 =根据时间间隔从不滚动）
    #a1.sinks.k1.hdfs.rollInterval = 7200   
    
    # 默认为10，滚动之前写入文件的事件数（0 =基于事件数从不滚动）
    a1.sinks.k1.hdfs.rollCount = 0     
    
    a1.sinks.k1.hdfs.filePrefix = %Y%m%d_
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    #————————————————————  绑定  ————————————————————#
    
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

备注：通过正则匹配kafka日志中的数据，将数据发送到不同的文件夹下，但是正在写入时文件格式为tmp类型，hive不能识别出文件，所以可以按2小时形成一个正式文件。当没有数据写入时，就不会生成tmp文件。

## 2.2flume切割文件相关参数 ##

#————————————————————  滚动切换：控制写文件的切换规则  ————————————————————#
    # 按文件体积（字节）来切
    ag1.sinks.sink1.hdfs.rollSize = 512000
    # 按event条数切
    ag1.sinks.sink1.hdfs.rollCount = 1000000 
    # 按时间间隔切换文件,创建文件多久后
    ag1.sinks.sink1.hdfs.rollInterval = 60 
    
    
    
    #————————————————————  时间切换：控制生成目录的规则  ————————————————————#
    
    ag1.sinks.sink1.hdfs.path =hdfs://hadoop04:9000/flume/%y/%m/%d/%H
    
    ag1.sinks.sink1.hdfs.round = true
    ag1.sinks.sink1.hdfs.roundValue = 10
    ag1.sinks.sink1.hdfs.roundUnit = minute
    ag1.sinks.sink1.hdfs.useLocalTimeStamp = true

# 三、Es数据采集 #

## 3.1采集mysql数据到Es ##

stu.conf文件

input {
      jdbc{
        # mysql 数据库链接
        jdbc_connection_string => "jdbc:mysql://localhost:3306/minisns?useUnicode=true&characterEncoding=utf-8&useSSL=false"
        # 用户名和密码
        jdbc_user => "123456"
        jdbc_password => "123456"
        #驱动
        jdbc_driver_library => "/opt/mysql/mysql-connector-java.jar"
        # 驱动类名
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_paging_enabled => "true"
        #jdbc_page_size => "5000"
        jdbc_page_size => "50000"
        jdbc_default_timezone =>"Asia/Shanghai"
        # mysql文件, 也可以直接写SQL语句在此处，如下：
        statement_filepath => "/opt/logstash/config/jdbc_sql/stu.sql"
        # 这里类似crontab,可以定制定时操作，比如每分钟执行一次同步(分 时 天 月 年)
        schedule => "*/5 * * * *"
        #type => "jdbc"
        # 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
        record_last_run => true
    
        # 是否需要记录某个column 的值,如果record_last_run为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
        use_column_value => true
        # 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是mysql主键
        tracking_column => "addtime"
        tracking_column_type => "datatime"
        last_run_metadata_path => "/opt/logstash/config/track_value/stu"
        # 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
        clean_run => false
        # 是否将 字段(column) 名称转小写
        lowercase_column_names => false
      }
    }
              
    output { 
      elasticsearch {
        hosts => ["http://es-01:9200","http://es-02:9200","http://es-03:9200"]
        index => "stu_v1"
        document_id => "%{Id}"
        user => "elastic"
        password => "123456"
      } 
    }

stu.sql文件

select
    Id
    ,name
    ,date_format(addtime, '%Y-%m-%d %H:%i:%s') as addtime
    from student
    where addtime >= date_add(:sql_last_value, interval -8 hour)

## 3.2采集Kafka数据到Es ##

#=========================== Filebeat inputs =============================
    
    filebeat.inputs:
    
    # Each - is an input. Most options can be set at the input level, so
    # you can use different inputs for various configurations.
    # Below are the input specific configurations.
    
    - type: kafka
      hosts:
        - kafka-01:9092
        - kafka-02:9092
        - kafka-03:9092
      topics: ["stu"]
      group_id: "test_group"
      # Change to true to enable this input configuration.
      #enabled: false
      
      
    #==================== Elasticsearch template setting ==========================
    
    setup.template.settings:
      index.number_of_shards: 3
      index.number_of_replicas: 1
      #index.codec: best_compression
      #_source.enabled: false
      
      
    #================================ Outputs =====================================
    
    # Configure what output to use when sending the data collected by the beat.
    
    #-------------------------- Elasticsearch output ------------------------------
    output.elasticsearch:
      hosts: ["es-01:9200","es-02:9200","es-03:9200"]
      username: "elastic"
      password: "123456"
      index: "student_%{+yyyyMM}"
    
    setup.template.name: "student_"
    setup.template.pattern: "student_*"
    setup.ilm.enabled: false
    
    
    #================================ Processors =====================================
    
    # Configure processors to enhance or manipulate events generated by the beat.
    
    # This allows to enable 6.7 migration aliases
    #migration.6_to_7.enabled: true
    processors:
      - drop_fields:
            fields: ['host','agent','ecs','kafka']
      - decode_json_fields:
            fields: ["message"]
            process_array: true
            max_depth: 1
            target: ""
            overwrite_keys: false
            add_error_key: true

## 3.3Logstash与FileBeat的对比 ##

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>logstash</td> 
   <td>filebeat</td> 
  </tr> 
  <tr> 
   <td>语言</td> 
   <td>Jruby</td> 
   <td>golang</td> 
  </tr> 
  <tr> 
   <td>应用场景</td> 
   <td>采集kafka数据后过滤分析发送到es</td> 
   <td>采集日志，然后发送到消息队列</td> 
  </tr> 
  <tr> 
   <td>优势</td> 
   <td>丰富的input|filter|output插件</td> 
   <td>轻量级</td> 
  </tr> 
 </tbody> 
</table>