Flink-StreaimingFileSink-自定义序列化-Parquet批量压缩

落日映苍穹つ 2023-07-12 14:41 18阅读 0赞

# Flink-StreaimingFileSink-自定义序列化-Parquet批量压缩 #

# Flink系列文章 #

*  更多Flink系列文章请点击[Flink系列文章][Flink]
 *  更多大数据文章请点击[大数据好文推荐][Link 1]

# 1 Maven依赖 #

Flink有内置方法可用于为Avro数据创建Parquet writer factory。

要使用ParquetBulkEncoder，需要添加以下Maven依赖：

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-parquet_2.11</artifactId>
      <version>1.10.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
        <version>1.10.0</version>
    </dependency>
    
    <dependency>
    	<groupId>org.apache.avro</groupId>
    	<artifactId>avro</artifactId>
    	<version>1.8.2</version>
    </dependency>
    
    <dependency>
    	<groupId>org.apache.parquet</groupId>
    	<artifactId>parquet-avro</artifactId>
    	<exclusions>
    		<exclusion>
    			<groupId>org.apache.hadoop</groupId>
    			<artifactId>hadoop-client</artifactId>
    		</exclusion>
    		<exclusion>
    			<groupId>it.unimi.dsi</groupId>
    			<artifactId>fastutil</artifactId>
    		</exclusion>
    	</exclusions>
    	<version>1.10.0</version>
    </dependency>

# 2 代码 #

## 2.1 main ##

从Kafka读数据，写入Parquet文件，并采用LZO压缩例子：

import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
    import org.apache.flink.formats.parquet.avro.ParquetAvroWriters
    import org.apache.avro.Schema
    
      def main(args: Array[String]) { 
     	// read parameter from command line
        val parameter = ParameterTool.fromArgs(args)
    
        // 1. set up the streaming execution environment
        val env = StreamExecutionEnvironment.getExecutionEnvironment
          // checkpoint every 5 minute
          .enableCheckpointing(5 * 60 * 1000)
    	  .setStateBackend(new RocksDBStateBackend(path, true))
    	  
        val checkpointConfig = env.getCheckpointConfig
        checkpointConfig.setMinPauseBetweenCheckpoints(2 * 60 * 1000)
        checkpointConfig.setCheckpointTimeout(3 * 60 * 1000)
        checkpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    
        // 2. kafka consumer`s config
        val kafkaConsumerConfig = new Properties()
        kafkaConsumerConfig.setProperty("bootstrap.servers", parameter.get("bootstrap-servers","192.168.1.1:9092"))
        kafkaConsumerConfig.setProperty("group.id", parameter.get("groupid","Kafka2hive"))
        kafkaConsumerConfig.setProperty("auto.offset.reset", parameter.get("offset","latest"))
        // 动态分区默认关闭，需要设置,以下表示10秒探测一次
        kafkaConsumerConfig.setProperty("flink.partition-discovery.interval-millis", "10000")
    
        // 3. create a kafka consumer
        val kafkaConsumer = new FlinkKafkaConsumer(
          "test_topic",
          new JSONKeyValueSilentDeserializationSchema,
          kafkaConsumerConfig)
        // 设置为用groupId放在ZK的offset，如果找不到就用配置文件中的auto.offset.reset来决定
        // 注意，该方法不影响此行为：从checkpoint/savepoint恢复时用存在这里面的offset消费
        .setStartFromGroupOffsets()
        // 1.9.0版本必须设这个值，虽然默认true，但是实测中并未提交offset到kafka
        // 设为true后已经观察到offset正常提交到kafka，便于监控
        .setCommitOffsetsOnCheckpoints(true)
    
        // 4. create the stream with kafka source, test_topic must return Student!
        val kafkaStream: DataStream[Student] = env
          .addSource(kafkaConsumer)
    
    	 // 5. 构建StreamingFileSink，指定BasePath和序列化Encoder
    	val sink: StreamingFileSink[Student] = StreamingFileSink
    	    .forBulkFormat(outputBasePath, ParquetAvroCompressionWriters.forReflectRecord(classOf[Student], CompressionCodecName.LZO))
    	    .withBucketAssigner(new EventDateTimeBucketAssigner("yyyyMMdd"))
    	    .build()
    	
    	// 6. 添加Sink到InputDataSteam即可
        kafkaStream.addSink(sink)
    
        // 7. execute program
        env.execute("Kafka to Parquet")
      }

## 2.2 JSONKeyValueSilentDeserializationSchema ##

该类十分重要。他继承自`KafkaDeserializationSchema`（他可用来访问Kafka消息的key、value和元数据。），负责解析FlinkKafkaConsumer从Kafka中读取到的`ConsumerRecord`。具体来说，会调用其`deserialize`方法。我们可以自定义处理逻辑，将其转为一个特定类型，再交由下一个算子处理。这里，我们是解析为了一个自定义Java Bean再由`StreamingFileSink`。

关于此类，有几点值得注意：

*  如果出现读Kafka数据中文乱码，可尝试按如下方式解析：

new String(record.value, StandardCharsets.UTF_8)

*  一定要做异常处理，出错时务必返回`null`来跳过该条异常记录。否则会造成报错=>程序重启=>继续消费这条记录=>再次报错=>再次重启。。。的死循环中。

## 2.3 ParquetAvroCompressionWriters ##

ParquetAvroCompressionWriters是以Avro格式来定义Parquet元数据，写入Parquet文件。

*  `ParquetAvroCompressionWriters.forReflectRecord(classOf[Student])`  
    以`Student`类来生成该Avro的schema，用来生成Parquet的元数据。

## 2.4 AvroSchema定义 ##

我们以Avro格式来定义Parquet元数据，写入Parquet文件。这里我们可以自定义类SensorPojo：

class SensorPojo extends Serializable{ 
      @BeanProperty @Nullable var kafka_partition:java.lang.Integer = _
      @BeanProperty @Nullable var kafka_offset:java.lang.Long = _
      @BeanProperty @Nullable var name:java.lang.Integer = _
      @BeanProperty @Nullable var age:String = _
      @BeanProperty @Nullable var time:java.lang.Long = _
    }

重点如下：

*  Serializable  
    定义类时，请`extends Serializable`。
 *  @Nullable  
    如果有字段为空，务必记得使用来自Avro项目的`@Nullable`注解允许有空值，否则遇到空值将会报错！
 *  如果使用scala，可以使用@BeanProperty字段注解，生成java风格的`get`、`set`方法。
 *  如果要写入Avro以外的Parquet兼容的数据格式，请实现[ParquetBuilder][]接口来创建ParquetWriterFactory。
 *  关于Parquet更多内容，可参考:
    
     *  [HDFS-Parquet][]

## 2.5 Hive-Parquet ##

如果要配合Hive使用，则在建表时直接指定`STORED AS parquet`即可（具体看hive版本）

## 2.6 Bucket ##

Bucket为按`yyyyMMdd`天格式生成，可以和Hive表分区目录配合。而且这里我们是自定义的`EventDateTimeBucketAssigner`，其他都跟`BucketAssigner`相同除了`getBucketId`方法自己实现根据特定字段获取BucketId:

public class EventDateTimeBucketAssigner implements BucketAssigner<Student, String> { 
    	@Override
        public String getBucketId(Student element, Context context) { 
            if (dateTimeFormatter == null) { 
                dateTimeFormatter = DateTimeFormatter.ofPattern(formatString).withZone(zoneId);
            }
            try { 
            	// Student类中的特定字段的get方法
                Long eventTime = element.getTime();
                if(eventTime != null && eventTime > 0){ 
                    return dateTimeFormatter.format(Instant.ofEpochMilli(eventTime));
                }
            } catch (Exception e) { 
                LOGGER.error("an error happened while dateTimeFormatter.format context.timestamp():", e);
            }
            return dateTimeFormatter.format(Instant.ofEpochMilli(context.currentProcessingTime()));
        }
    }

# 3 总结 #

## 3.1 关于压缩 ##

可参考:

*  [大数据-常用压缩方式总结][-]

这里采用LZO，需要建立index才能对MR任务切分。

我们实测中，采用flume+LZO方式写文件和Flink+Parquet方式相比，总大小差距不大，所以必须在Parquet基础上加上LZO。

## 3.2 小文件 ##

### 3.2.1 概述 ###

因为每个subtask在一个checkpoint周期就会生成一个文件，所以在并发高时小文件数量很大，不仅增加NameNode维护元数据成本，也影响下游其他任务读取效率（大量小文件大量磁盘IO）。常见调优方式介绍如下。

### 3.2.2 增加Checkpoint周期 ###

因为使用BulkEncoding时只能用`OnCheckpointRollingPolicy`，所以我们调大Checkpoint间隔可以减少总的part-file文件数量。

但调大以后，会增加每次Checkpoint时间，以及增长数据可见周期，需要权衡。