传统的流式数据入库的过程中对小文件进行合并会产生很多问题，比如流式数据不断的往hive表进行写入，如果同时有一个合并程序进行小文件的合并，那么这时候对同一份数据进行读写。会不会产生问题。如何保证事务，出错了怎么回滚呢，这些都是很棘手的问题。我们的流任务以flink为主，查询引擎是presto，所以调研以后，我决定引入iceberg来解决小文件合并的问题。

在flink 1.11中，内置提供了一个DataGen 连接器，主要是用于生成一些随机数，用于在没有数据源的时候，进行流任务的测试以及性能测试等。下面我们简单的聊聊如何来使用以及底层源码是如何实现的。

在flink中，StreamingFileSink是一个很重要的把流式数据写入文件系统的sink，可以支持写入行格式(json,csv等)的数据，以及列格式（orc、parquet）的数据。hive作为一个广泛的数据存储，而ORC作为hive经过特殊优化的列式存储格式，在hive的存储格式中占有很重要的地位。今天我们主要讲一下使用StreamingFileSink将流式数据以ORC的格式写入hdf

今天我们主要来讲一个很简单但是很常见的需求，实时计算出网站当天的pv值，然后将结果实时更新到mysql数据库，以供前端查询显示。接下来我们看看如何用flinksql来实现这个简单的功能。首先我们还是使用datagen生成测试数据，随机生成一些用户idString sourceSql = "CREATE TABLE datagen (\n" +" userid int,\n" +" proctime

其实我们可以想一下这个工作大概是什么流程，首先要写入hive，我们首先要从hive的元数据里拿到相关的hive表的信息，比如存储的路径是哪里，以便往那个目录写数据，还有存储的格式是什么，orc还是parquet，这样我们需要调用对应的实现类来进行写入，其次这个表是否是分区表，写入数据是动态分区还是静态分区，这些都会根据场景的不同而选择不同的写入策略。写入数据的时候肯定不会把所有数据写入一个文件，那

文章目录背景iceberg简介flink实时写入准备sql client环境创建catalog创建db创建table插入数据查询代码版本总结背景随着大数据处理结果的实时性要求越来越高，越来越多的大数据处理从离线转到了实时，其中以flink为主的实时计算在大数据处理中占有重要地位。Flink消费kafka等实时数据流。然后实时写入hive，在大数据处理方面有着广泛的应用。此外由于列式存储格式如par

1.11.0 版本后，用户使用 Flink SQL 时可以自动获取表的 schema 而不再需要输入 DDL。除此之外，任何 schema 不匹配的错误都会在编译阶段提前进行检查报错，避免了之前运行时报错造成的作业失败。

目前对于flink来说，生产环境一般有两个部署模式，一个是 session模式，一个是per job模式。session模式共享资源，如果一个任务出现了问题，会影响其他的任务，per job模式需要有大量的jar和资源拷贝，并且生成JobGraph都是在本地做的，如果任务多的话，会给服务器造成很大的压力，所以flink引入了application模式。Yarn Application 模式会在客户

flink 1.11 支持用户直接使用sql将流式数据写入hive，并且可以自动的创建和刷新hive的分区，支持的数据格式包括json、csv、parquet、csv。底层是使用了写入文件系统的功能，所以具体的配置可以参考写入文件系统的配置。

文章目录背景flink broadstate简介使用方法实例讲解简述一下需求自定义source背景紧接着上一篇文章【FLINK实战-使用CEP进行网站监控报警和报警恢复】，上一篇主要讲了怎么对数据流做监控报警，但是实际生产环境中，可能我们对这个报警的阈值设置成多大也没有一个准，可能需要根据经验不断的来修改，所以就涉及了可能需要不断的修改这个报警的阈值，但是如果每次修改了之后，都通过重启flink程