Flink
未读6. Flink 时间语义6.1 三种时间概念在实时流式计算中，”时间”是一个能影响计算结果的非常重要因素！试想场景：每隔1分钟计算一次最近10分钟的活跃用户量； 假设此刻的时间是13:10，要计算的活跃用户量时间段为：[13:00, 13:10） 有一条行为日志中记录的用户行为时间是12:59，但到达Flink的计算程序已经是13:02，那么。这个用户是否要纳入本次计算的结果中呢？看如何定义： 如果时段[13:00, 13:10)定义的是用户行为的发生时间(数据中的业务时间)，则不应纳入； 如果时段[13:00, 13:10)定义的是计算时的时间，则应该纳入； Flink内部为了直观地统一计算时所用的时间标准，特制定了两种时间语义： processing time 处理时间 event time 事件时间 Ingestion time 注入时间 时间语义注意影响”窗口计算”； 6.1 两种时间语义时间语义，是Flink中用于时间推进和时间判断的机制；时间推进和时间判断，以什么为标准，就产出了两种不同的时间语义； 以processing time为依据， ...
Flink
未读5. Flink 并行度相关API5.1 基础概念 用户通过算子API所开发的代码，会被Flink任务提交客户端解析成jobGraph 然后，jobGraph提交到集群JobManager，转化为ExecutionGraph(并行化后的执行图) 然后，ExecutionGraph中的各个task会以多并行实例(subTask)部署到taskManager上执行； subTask运行的位置时taskManager所提供的槽位(task slot)，槽位简单理解就是线程； 重要提示 一个算子的逻辑，可以封装在一个独立的task中(可以有多个运行时实例：subTask)； 也可以把多个算子的逻辑chain在一起后封装在一个独立的task中(可以有多个运行时实例：subTask) 同一个task的多个并行实例，不能放在同一个taskslot上运行。一个taskslot，可以运行多个不同task的各自1个并行实例。job中只要有一个task的并行度超过集群可用的总槽位数，这个job就会提交失败。所谓的task，就是对算子的封装，用户写的这些算子是用来表达计算逻辑的，本身是不能被执行的，真 ...
4. Flink 编程 process function4.1 process function概述process function相对于前文所述的map、flatMap、filter算子来说，最大的区别就是其让开发人员对数据的处理逻辑拥有更大的自由度；process function把数据及必要的上下文提供给开发人员，然后如何处理，如何返回，全部交给用户来控制；在事件驱动型的应用中，使用最频繁的api往往就是process function；在不同类型的DataStream上，(比如keyedStream、windowedStream、ConnectedStream等)，应用process function时，flink提供了大量不同类型的process function，让其针对不同的DataStream拥有更具针对性的功能。 在不同类型的流上，调用process算子，需要传入的ProcessFunction也不同。ProcessFunction继承RichFunction，可以使用生命周期方法。 各种算子运算后生成的DataStream类型，及各种DataStream类型之间的相 ...
Flink
未读3. Flink 多流操作API3.1 split分流操作[已deprecated]split拆分 (DataStream -> SplitStream) 该方法是将一个DataStream中的数据流打上不同的标签，逻辑的拆分成多个不同类型的流，返回一个新的SplitStream，本质上还是一个数据流，只不过是将流中的数据打上了不同的标签。 返回的SplitStream已经标记为过时，1.12版本已经删除了split和select方法，可以使用测流输出代替。 3.2 分流操作[使用测流输出]123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051public class _13_SideOutput_Demo { public static void main(String[] args) throws Exception { Configuration configuration = new Configurati ...
Flink
未读2.9 Flink 扩展sink算子2.9.1 KafkaSinkFlink可以和Kafka多个版本整合，比如0.11.x、1.x、2.x等；从Flink1.9开始，使用的是kafka2.2的客户端。 核心类 KafkaStringSerializationSchema – 反序列化 FlinkKafkaProducer – 生产者（即sink） 需要添加依赖： 12345<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>1.14.4</version></dependency> 新版本API 12345678910111213141516171819202122232425262728293031323334public class _10_KafkaSinkOperator_Demo1 & ...
Flink
未读2.8 Flink 基础sink算子2.8.1 打印输出print打印是最简单的一个Sink，通常是用来做实验和测试时使用。 1SingleOutputStreamOperator<Tuple2<String, Integer>> resultStream = keyed.sum("f1"); 2.8.2 文件Sink以下writeAs... 方法均已被标记为deprecated writeAsText 以文本格式输出 该方法是将数据以csv格式写入到指定的目录中，目录中的文件名称是该Sink所在subtask的Index+1。可以额外指定一个参数writeMode，默认是WriteMode.NO_OVERWRITE。如果是WriteMode.OVERWRITE，会将以前的文件覆盖。 writeAsCsv 以csv格式输出 该方法是将数据以csv格式写入到指定的目录中，本质上使用的是CsvOutputFormat格式写入的。该Sink并不是将数据实时写入到文件中，而是有一个BufferedOutputStream，默认缓存的大小为40 ...
2.7 Flink 基础transformation算子2.7.1 映射算子map 映射（DataStream -> DataStream） map(new MapFunction)MapFunction : (x) -> y [1条变1条] 12345678910111213StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> words = env.fromElements("hadoop", "spark", "flink", "hbase", "flink");//在map方法中传入MapFunction实现类实例，重写map方法SingleOutputStreamOperator<String> upperWords = words.map(new MapFunction<S ...
2.6 Flink 基础source算子source 是用来获取外部数据的算子，按照获取数据的方式，可以分为： 基于集合的Source 基于Socket网络端口的Source 基于文件的Source 第三方Connector Source 自定义Source 从并行的角度，source又可分为非并行的source和并行的source 非并行source：并行度只能为1，即只有一个运行时实例，在读取大量数据时效率比较低，通常是用来做一些实验或测试，例如Socket Source。 并行的source：并行度可以是1到多个，在计算资源足够的前提下，并行度越大，效率越高，例如Kfaka Source。 2.6.1 基于集合的Source（测试用）可将一个普通的Java集合、迭代器或者可变参数转换成一个分布式数据流DataStream fromElements 1234567891011public class SourceOperator { public static void main(String[] args) throws Exception & ...
Flink
未读2. Flink编程基础2.1 需要引入的基本依赖123456789101112131415<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.14.4</version></dependency><dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>1.14.4</version></dependency><dependency> <groupId>org.apache.flink</ ...
Spark
未读云服务器伪分布式部署Spark1.下载Spark 下载地址安装spark前需要安装scala （scala下载地址） 2.上传至服务器并解压这里我解压到/home/software/目录下 1tar -zxvf spark-3.1.2-bin-hadoop3.2.tgz -C /home/software/ 3.配置环境变量1vim /etc/profile 添加如下 123#这里没有添加sbin目录，因为会和hadoop的sbin目录冲突export SPARK_HOME=/home/software/spark-3.1.2-bin-hadoop3.2export PATH=$PATH:$SPARK_HOME/bin 4.配置conf4.1 配置spark-env.sh进入conf目录 1cd /home/software/spark-3.1.2-bin-hadoop3.2/conf/ 复制spark-env.sh.template，重命名为spark-env.sh 1cp spark-env.sh.template spark-env.sh ...
