Flink落盘Parquet文件

2019年3月2日 0 条评论 4.22k 次阅读 2 人点赞

前言

在实时数仓方面除了直接将数据写入Druid，Kafka等相关的消息队列以外，Flink还支持将数据写出到HDFS上并以Parquet格式进行存储，本案例简单地介绍了整个过程的实现，为后续实时数仓或者其他业务线提供参考。

依赖

除了Flink所依赖的常规Jar包以外，还需要提供以下依赖：

Scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0-cdh5.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parquet</artifactId>
<version>1.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId>
<version>1.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-avro</artifactId>
<version>1.10.0</version>
</dependency>

编码

Flink同时支持Java和Scala，本文中采用Scala进行示例：

Scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
package com.yidian.cjv
 
 
import com.alibaba.fastjson.JSON
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.core.fs.Path
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala._
 
import scala.collection.mutable.ListBuffer
import scala.util.Random
 
/**
  * @author: w3sun
  * @date: 2019/3/3 20:49
  * @description:
  *
  */
object String2Parquet {
  //模拟生成数据
  def generateSource(out: SourceContext[String]): Unit = {
    val lb = new ListBuffer[String]
    lb += "{\"time\":\"2018-12-10\",\"id\":\"1\",\"content\":\"realtime\"}"
    lb += "{\"time\":\"2019-12-10\",\"id\":\"2\",\"content\":\"batch\"}"
    lb += "{\"time\":\"2020-12-10\",\"id\":\"3\",\"content\":\"so what\"}"
    while (true) {
      val index = Random.nextInt(3)
      Thread.sleep(200)
      out.collect(lb(index))
    }
  }
 
  def main(args: Array[String]): Unit = {
    //参数校验
    val params = ParameterTool.fromArgs(args)
    val output = params.get("output")
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE)
    val timeAssigner = new DateTimeBucketAssigner[LogEvent]("yyyy-MM-dd-HH-mm")
    val source: DataStream[LogEvent] = env.addSource(generateSource _)
      .map(line => {
        val event: LogEvent = JSON.parseObject(line, classOf[LogEvent])
        event
      })
    val sink = StreamingFileSink.forBulkFormat(new Path(output),
      ParquetAvroWriters.forReflectRecord(classOf[LogEvent])).withBucketAssigner(timeAssigner).build()
    source.addSink(sink).name("String2Parquet")
    env.execute(this.getClass.getSimpleName)
  }
 
  //case class 默认实现Serializable
  case class LogEvent(time: String, id: String, content: String) {
    override def toString: String = time + "\t" + id + "\t" + content
  }
}

效果

1
2
3
4
5
6
7
total 0
drwxr-xr-x    6 neters  staff   192 Mar  2 17:57 .
drwxr-xr-x    4 neters  staff   128 Mar  2 17:54 ..
drwxr-xr-x   16 neters  staff   512 Mar  2 17:55 2019-03-02-17-54
drwxr-xr-x  242 neters  staff  7744 Mar  2 17:56 2019-03-02-17-55
drwxr-xr-x  244 neters  staff  7808 Mar  2 17:57 2019-03-02-17-56
drwxr-xr-x   25 neters  staff   800 Mar  2 17:57 2019-03-02-17-57

1
2
3
4
5
6
7
8
9
10
11
12
13
14
-rw-r--r--   1 neters  staff  784 Mar  2 17:54 part-0-0
-rw-r--r--   1 neters  staff  838 Mar  2 17:54 part-0-1
-rw-r--r--   1 neters  staff  838 Mar  2 17:54 part-0-2
-rw-r--r--   1 neters  staff  867 Mar  2 17:55 part-0-3
-rw-r--r--   1 neters  staff  867 Mar  2 17:54 part-1-0
-rw-r--r--   1 neters  staff  787 Mar  2 17:54 part-1-1
-rw-r--r--   1 neters  staff  838 Mar  2 17:54 part-1-2
-rw-r--r--   1 neters  staff  867 Mar  2 17:54 part-2-0
-rw-r--r--   1 neters  staff  867 Mar  2 17:54 part-2-1
-rw-r--r--   1 neters  staff  784 Mar  2 17:54 part-2-2
-rw-r--r--   1 neters  staff  838 Mar  2 17:54 part-3-0
-rw-r--r--   1 neters  staff  858 Mar  2 17:54 part-3-1
-rw-r--r--   1 neters  staff  867 Mar  2 17:54 part-3-2
-rw-r--r--   1 neters  staff  867 Mar  2 17:55 part-3-3

可以看到，数据按照分钟级别生成文件夹，然后再生成多个文件。

注意事项

Streaming到Parquet数据的生成是由Checkpoint触发的，因此必须设置Checkpoint为Enable状态，至于数据刷盘时间根据业务线数据量不同自行决定。

© 著作权归作者所有

勇敢，和生活的艰难无关。

文章评论（0）