1. 云图网首页
  2. 技术专区
  3. 大数据

Flink生成Parquet格式文件实战详解大数据

• 大数据

1.概述

在流数据应用场景中，往往会通过Flink消费Kafka中的数据，然后将这些数据进行结构化到HDFS上，再通过Hive加载这些文件供后续业务分析。今天笔者为大家分析如何使用Flink消费Kafka的数据后，将消费后的数据结构化到Hive数据仓库中。

2.内容

Hive能够识别很多类型的文件，其中包含Parquet文件格式。因此，我们只需要将Flink消费Kafka后的数据以Parquet文件格式生成到HDFS上，后续Hive就可以将这些Parquet文件加载到数据仓库中。具体流程图如下所示：

Flink生成Parquet格式文件实战详解大数据

2.1 Flink On YARN

 实现整个案例，我们需要Hadoop环境、Kafka环境、Flink环境、Hive环境。这里，笔者只介绍Flink环境的部署，其他环境可自行搜索部署方案。关于Flink On YARN的安装步骤如下：

2.1.1 准备安装包

【官方下载地址】

2.2.2 解压

解压命令如下所示：

# 解压Flink安装包并重名名为flink 
tar -zxvf flink-1.7.1-bin-hadoop27-scala_2.12.tgz && mv flink-1.7.1 flink 
# 配置环境变量 
vi ~/.bash_profile 
 
# 添加如下内容 
export FLINK_HOME=/data/soft/new/flink 
export PATH=$PATH:$FLINK_HOME/bin 
 
# 保存并退出

Flink On YARN有两种模式，分别是Flink Session和Flink Job On YARN。

2.2.3 Flink Session

Flink Session命令如下：

# 启动一个Flink Session 
yarn-session.sh -n 2 -jm 1024 -tm 1024 -d

各个参数含义如下：

参数 含义
-n 2 指定2个容器
-jm 1024 JobManager内存为1024MB
-tm 1024 TaskManager内存为1024MB
-d 任务后台运行

如果你不想让Flink YARN客户端一直运行，也可以启动分离的YARN Session，通过参数-d来实现。这种情况下Flink YARN客户端只会将Flink提交给集群，然后自行关闭。需要注意的是，这种情况无法使用Flink停止YARN会话，需要使用YARN的命令来停止，命令如下：

yarn application -kill <appId>

2.2.4 Flink On YARN

命令如下：

# yarn-cluster模式提交Flink任务 
flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 WordCount.jar

各个参数含义如下：

参数 含义
-m yarn-cluster 连接指定集群，如使用标识yarn-cluster
-yn 2 2个容器
-yjm 1024 JobManager内存为1024MB
-ytm TaskManager内存为1024MB

如果不知道提交队列，任务会提交到默认队列中，如果需要指定提交队列，可以使用参数-yqu queue_01进行提交。

3.消费Kafka并生成Parquet文件

准备一个Topic的Schema类TopicSource，TopicSource类定义如下：

public class TopicSource { 
 
    private long time; 
    private String id; 
 
    public long getTime() { 
        return time; 
    } 
 
    public void setTime(long time) { 
        this.time = time; 
    } 
 
    public String getId() { 
        return id; 
    } 
 
    public void setId(String id) { 
        this.id = id; 
    } 
 
}

编写一个生成Parquet的Flink类FlinkParquetUtils，具体代码实现如下：

/** 
* Consumer kafka topic & convert data to parquet. 
*  
* @author smartloli. 
* 
*         Created by Feb 24, 2019 
*/ 
public class FlinkParquetUtils { 
private final static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
private final static Properties props = new Properties(); 
static { 
/** Set flink env info. */ 
env.enableCheckpointing(60 * 1000); 
env.setParallelism(1); 
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
/** Set kafka broker info. */ 
props.setProperty("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092"); 
props.setProperty("group.id", "flink_group_parquet"); 
props.setProperty("kafka.topic", "flink_parquet_topic_d"); 
/** Set hdfs info. */ 
props.setProperty("hdfs.path", "hdfs://cluster1/flink/parquet"); 
props.setProperty("hdfs.path.date.format", "yyyy-MM-dd"); 
props.setProperty("hdfs.path.date.zone", "Asia/Shanghai"); 
props.setProperty("window.time.second", "60"); 
} 
/** Consumer topic data && parse to hdfs. */ 
public static void getTopicToHdfsByParquet(StreamExecutionEnvironment env, Properties props) { 
try { 
String topic = props.getProperty("kafka.topic"); 
String path = props.getProperty("hdfs.path"); 
String pathFormat = props.getProperty("hdfs.path.date.format"); 
String zone = props.getProperty("hdfs.path.date.zone"); 
Long windowTime = Long.valueOf(props.getProperty("window.time.second")); 
FlinkKafkaConsumer010<String> flinkKafkaConsumer010 = new FlinkKafkaConsumer010<>(topic, new SimpleStringSchema(), props); 
KeyedStream<TopicSource, String> KeyedStream = env.addSource(flinkKafkaConsumer010).map(FlinkParquetUtils::transformData).assignTimestampsAndWatermarks(new CustomWatermarks<TopicSource>()).keyBy(TopicSource::getId); 
DataStream<TopicSource> output = KeyedStream.window(TumblingEventTimeWindows.of(Time.seconds(windowTime))).apply(new WindowFunction<TopicSource, TopicSource, String, TimeWindow>() { 
/** 
*  
*/ 
private static final long serialVersionUID = 1L; 
@Override 
public void apply(String key, TimeWindow timeWindow, Iterable<TopicSource> iterable, Collector<TopicSource> collector) throws Exception { 
iterable.forEach(collector::collect); 
} 
}); 
// Send hdfs by parquet 
DateTimeBucketAssigner<TopicSource> bucketAssigner = new DateTimeBucketAssigner<>(pathFormat, ZoneId.of(zone)); 
StreamingFileSink<TopicSource> streamingFileSink = StreamingFileSink.forBulkFormat(new Path(path), ParquetAvroWriters.forReflectRecord(TopicSource.class)).withBucketAssigner(bucketAssigner).build(); 
output.addSink(streamingFileSink).name("Sink To HDFS"); 
env.execute("TopicData"); 
} catch (Exception ex) { 
ex.printStackTrace(); 
} 
} 
private static TopicSource transformData(String data) { 
if (data != null && !data.isEmpty()) { 
JSONObject value = JSON.parseObject(data); 
TopicSource topic = new TopicSource(); 
topic.setId(value.getString("id")); 
topic.setTime(value.getLong("time")); 
return topic; 
} else { 
return new TopicSource(); 
} 
} 
private static class CustomWatermarks<T> implements AssignerWithPunctuatedWatermarks<TopicSource> { 
/** 
*  
*/ 
private static final long serialVersionUID = 1L; 
private Long cuurentTime = 0L; 
@Nullable 
@Override 
public Watermark checkAndGetNextWatermark(TopicSource topic, long l) { 
return new Watermark(cuurentTime); 
} 
@Override 
public long extractTimestamp(TopicSource topic, long l) { 
Long time = topic.getTime(); 
cuurentTime = Math.max(time, cuurentTime); 
return time; 
} 
} 
public static void main(String[] args) { 
getTopicToHdfsByParquet(env, props); 
} 
}

然后将编写好的应用程序进行打包，这里我们可以利用Maven命令，可以很方便的进行打包，在pom.xml文件中添加如下插件：

<build> 
<plugins> 
<plugin> 
<groupId>org.apache.maven.plugins</groupId> 
<artifactId>maven-compiler-plugin</artifactId> 
<version>3.3</version> 
<configuration> 
<source>1.8</source> 
<target>1.8</target> 
</configuration> 
</plugin> 
<plugin> 
<artifactId>maven-assembly-plugin</artifactId> 
<configuration> 
<archive> 
<manifest> 
<mainClass>org.smartloli.kafka.connector.flink.hdfs.FlinkParquetUtils</mainClass> 
</manifest> 
</archive> 
<descriptorRefs> 
<descriptorRef>jar-with-dependencies</descriptorRef> 
</descriptorRefs> 
</configuration> 
</plugin> 
</plugins> 
</build>

然后使用如下命令进行编译打包：

mvn clean && mvn assembly:assembly

最后将打包的JAR上传到Flink集群。

4.运行及预览

将应用程序的JAR上传到Flink集群后，执行如下命令进行提交：

flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 -yqu hadoop kafka-connector-flink-parquet.jar

查看ResourceManager的页面，提交任务如下：

Flink生成Parquet格式文件实战详解大数据

 在代码中，我们在HDFS上以日期yyyy-MM-dd的格式进行生产，结果如下：

Flink生成Parquet格式文件实战详解大数据

5.总结

在编写Flink应用程序的时候，建议使用Maven来管理项目，这样添加依赖JAR的时候，只需将依赖的信息添加到pom.xml文件即可。打包的时候，同样使用Maven命令，这样应用程序所依赖的JAR包均会打包进行，防止遗漏导致提交任务时失败。

6.结束语

这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

另外，博主出书了《Kafka并不难学》，喜欢的朋友或同学， 可以在公告栏那里点击购买链接购买博主的书进行学习，在此感谢大家的支持。

原创文章，作者：奋斗，如若转载，请注明出处：https://blog.ytso.com/9866.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论