flink

首先安装flink,这里以1.13.1举例(docker安装的话,jobmanager和taskmanager需要分开部署)flink-docker-github-repository (这个参数flink文档中没找到)关于query.server.port: 6125不用去管,这个不填也会默认自动生成: 12yml中的内容不说了, 下面贴几张关于Flink生产配置最佳实践的图, 钉钉公开课趣头条实时平台负责人分享https://www.bilibili.com/video/BV1iE411r7S6 Flink在yarn上运行，每个TaskManager的slot个数怎么设置？经验公式：slot个数tm个数=并行度并行度=kafka的分区个数（10分区）slot的个数要小于yarn设置的单个container最大可以申请的cpu核数（5个 8-36个）。那么就是 5个slot2个tm=并行度=kafka分区数多个小集群好还是一个大集群好？1.小集群的话任务分布在不同的集群，任务之间的影响小。不会因为一个任务占用 ...

2022-01-20 09:25:58,308 WARN org.apache.flink.client.deployment.application.DetachedApplicationRunner [] - Could not execute application:org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Job was submitted in detached mode. Results of job execution, such as accumulators, runtime, etc. are not available. Please make sure your program doesn't call an eager execution function [collect, print, printToErr, count] 这是因为当 flink 作业是以 detached 模式提交的时候，作业不 ...

截至1.13.1,官方文档所提供的方式已经废弃12345678910111213141516171819202122232425262728293031HttpHost httpHost = new HttpHost(esHost, esPort, esScheme);List<HttpHost> httpPosts = new ArrayList<>();httpPosts.add(httpHost);RestClientFactory restClientFactory = new RestClientFactory() { @Override public void configureRestClientBuilder(RestClientBuilder restClientBuilder) { CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCrede ...

记一次使用coGroup产生的问题(Caused by: org.apache.flink.util.TraversableOnceException: The Iterable can be iterated over only once. Only the first call to 'iterator()' will succeed)

(flink)问题:esotericsoftware.kryo.serializers.CollectionSerializer.read NullPointerException 异常如下 123456789101112131415161718192021222324252627282930313233com.esotericsoftware.kryo.KryoException: java.lang.NullPointerExceptionSerialization trace:values (org.apache.avro.generic.GenericData$Record) at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125) at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528) at com.esotericsoftware.kr ...

12345678910111213/** * { * 1:[2,3,4], * 2:[3,4,5] * } * ↓ * { * 2:[1], * 3:[1,2], * 4:[1,2], * 5:[2] * } */ flink由上转成下面的数据格式 打散 1234562:1,3:1,4:1,3:2,4:2,5:2 2.再根据key分组并且reduce 12345678910111213141516171819202122232425262728DataSet<Tuple2<String, String>> test = ...test.flatMap(new FlatMapFunction<Tuple2<String, String>, Tuple2<String, String>>() { @Override public void flatMap(Tuple2<String, String> valu ...

clickhouse什么是clickhouse？优势是什么，为什么选择ch？ 全称 ：Click Stream Data WareHouse 简称ClickHouse 是一个用于联机分析(OLAP)的列式数据库管理系统(DBMS)。 ClickHouse的核心特性 ClickHouse拥有完备的管理功能，所以它称得上是一个DBMS（Database Management System，数据库管理系统），而不仅是一个数据库。 DDL（数据定义语言）：可以动态地创建、修改或删除数据库、表和视图，而无须重启服务。 DML（数据操作语言）：可以动态查询、插入、修改或删除数据。 权限控制：可以按照用户粒度设置数据库或者表的操作权限，保障数据的安全性。 数据备份与恢复：提供了数据备份导出与导入恢复机制，满足生产环境的要求。 分布式管理：提供集群模式，能够自动管理多个数据库节点。 列式存储与数据压缩 列式存储：想让查询变得更快，最简单且有效的方法是减少数据扫描范围和数据传输时的大小。假设一张数据表A拥有50个字段A1～A50，以及100行数据。现在需要查询前5个字段并进行数据分析，则可 ...

Flink什么是Flink？优势是什么，为什么选择Flink？ApacheFlink是一个框架和分布式处理引擎，用于在无界和有界数据流上进行有状态计算。Flink被设计为在所有通用集群环境中运行，以内存速度和任何规模执行计算。 经典的MapReduce程序 算子的概念：可以并行执行的方法称之为算子。 hadoop：在MR程序执行时，基本上每一次的数据交换都会读写磁盘。在复杂的业务逻辑下，由于执行的job任务多造成多次的磁盘读写，严重的影响了计算的时效性。这就是第一代计算框架hadoop的不足。 Tez：Tez是一个Hive的运行引擎，优化业务之间的读写磁盘次数，性能优于MR。 Spark：Spark的设计模式是读取集群中的数据后，在内存中存储和运算，直到全部运算完毕后，再存储到集群中。 Flink：在批处理的性能与Spark是差不多的，在底层设计的模式上Spark是以批处理为出发点，流是批处理的一种特例，Flink是以流处理为出发点，批处理是流的一种特例。 总结：在处理的的效率上MR的计算方式基本上被淘汰了。在批处理方面Spark&Flink性能是差不多的。 ...

异常:DataRowException, internal schema representation is probably out of sync with real database schema 123456789起初异常信息打印的不全, 但是浑然不知,后来在打印信息里发现flink并没有打印出应该打印的日志,然后去查了下,是缺少依赖包包加上<dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.25</version> <!-- 注意，若无type为jar则报错--> <type>jar</type></dependency> 全部异常信息 1234567891011121314151617181920212223242526272829303132333435ERROR io. ...

先查看centos中自带的jdk并卸载 1234567[root@root ~]# rpm -qa | grep java //查看tzdata-java-2016c-1.el6.noarchjava-1.6.0-openjdk-1.6.0.38-1.13.10.4.el6.x86_64java-1.7.0-openjdk-1.7.0.99-2.6.5.1.el6.x86_64[root@root ~]# rpm -e --allmatches --nodeps java-1.6.0-openjdk-1.6.0.38-1.13.10.4.el6.x86_64 //卸载[root@root ~]# rpm -e --allmatches --nodeps java-1.7.0-openjdk-1.7.0.99-2.6.5.1.el6.x86_64 //卸载[root@root ~]# rpm -qa | grep java //再次查看 开发版openjdk(非开发版还需单独额外安装jps等工具) 1yum install -y java-1.8.0-open ...