2022-05-07 c307bc4c93e71c200165f4a22988a191 99+ fast 0.1 k

流批选择

之前版本

//流
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//批
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

现在版本

通过执行模式 execution mode选择

1 流处理 streaming 默认

2 批处理 batch

3 自动 automatic

（1）通过命令行

1	flink run -Dexecution.runtime-mode=BATCH/../..

（2）代码

1
2
3

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
       env.setRuntimeMode(RuntimeExecutionMode.BATCH);
       env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

大数据基础组件 flink 使用

流批选择

2022-05-06 8f935ed29e76c7ff2a9dcdc7a339d376 99+ fast 0.1 k

hql增删改查

1增

insert

load

2删

1 表

Drop 表结构都没有了

1	DROP TABLE IF EXISTS employee;

2 记录

没有DELETE

TRUNCATE

所有记录

truncate table employees;

INSERT OVERWRITE

1 2	INSERT OVERWRITE TABLE dpc_test SELECT * FROM dpc_test WHERE age is not null;

3改

1 update

针对记录

1	update student set id='444' where name='tom';

2 Alter

表结构

4查

select

大数据基础组件 hive hql

hql增删改查

2022-05-05 7a9a85a9cc3acd423b6b9b69ca3f99cf 99+ 2 m 0.2 k

建表

CREATE TABLE IF NOT EXISTS `runoob_tbl`(
   `runoob_id` INT UNSIGNED AUTO_INCREMENT,
   `runoob_title` VARCHAR(100) NOT NULL,
   `runoob_author` VARCHAR(40) NOT NULL,
   `submission_date` DATE,
   PRIMARY KEY ( `runoob_id` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

字段数据类型

https://www.w3school.com.cn/sql/sql_datatypes.asp

array

https://www.educba.com/array-in-sql/

约束（Constraints）

NOT NULL - 指示某列不能存储 NULL 值。
UNIQUE - 保证某列的每行必须有唯一的值。
PRIMARY KEY - NOT NULL 和 UNIQUE 的结合。确保某列（或两个列多个列的结合）有唯一标识，有助于更容易更快速地找到表中的一个特定的记录。
FOREIGN KEY - 保证一个表中的数据匹配另一个表中的值的参照完整性。
CHECK - 保证列中的值符合指定的条件。
DEFAULT - 规定没有给列赋值时的默认值。

自增字段

AUTO INCREMENT

CREATE TABLE Persons
(
ID int NOT NULL AUTO_INCREMENT,
LastName varchar(255) NOT NULL,
FirstName varchar(255),
Address varchar(255),
City varchar(255),
PRIMARY KEY (ID)
)

开始值是 1，每条新记录递增 1

大数据基础组件数据库关系型 sql

建表

2022-04-25 d6d920f202b745618e6d05cf5def9373 99+ fast 0.0 k

任务生成和分配

main代码 -》数据流图（dataflow graph，logical streamgraph） -》作业图（jobgraph）-》执行图（executiongraph）-> 物理图（physical graph）

大数据基础组件 flink 原理

任务生成和分配

2022-04-25 6c9dc21d9f6a93a3b17e7ff45db3d999 99+ fast 0.1 k

任务槽 task slots

slot共享

并行度：算子的子任务个数

程序的并行度：最大算子并行度

假设：

设置全局并行度为6，保持sink为1

source，map(6) -》keyby。。。(6) -》sink(1)

总共有13个子任务

2 个taskmanger , 每个taskmanger 3个slot

大数据基础组件 flink 原理

任务槽 task slots

2022-04-25 8edcc8221b3c7e658d154f7a5ea48c50 99+ fast 0.1 k

算子链

多个算子合并

合并条件：1 并行度相同的算子 2 一对一 one to one

好处：1. 减少线程之间的切换和缓存区的数据交换 2 减少时延 3 提高吞吐量

大数据基础组件 flink 原理

算子链

2022-04-24 16f31f690b5a83132baf4757892eabbe 99+ 4 m 0.6 k

datastream

1 转换算子 Transformation

function分类：普通的，rich

怎么写function：

自定义
匿名类
lambda表达式

package com.atguigu.chapter05;

/**
 * Copyright (c) 2020-2030 尚硅谷 All Rights Reserved
 * <p>
 * Project:  FlinkTutorial
 * <p>
 * Created by  wushengran
 */

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransReturnTypeTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> clicks = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );

        // 想要转换成二元组类型，需要进行以下处理
        // 1) 使用显式的 ".returns(...)"
        DataStream<Tuple2<String, Long>> stream3 = clicks
                .map( event -> Tuple2.of(event.user, 1L) )
                .returns(Types.TUPLE(Types.STRING, Types.LONG));
        stream3.print();


        // 2) 使用类来替代Lambda表达式
        clicks.map(new MyTuple2Mapper())
                .print();

        // 3) 使用匿名类来代替Lambda表达式
        clicks.map(new MapFunction<Event, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Event value) throws Exception {
                return Tuple2.of(value.user, 1L);
            }
        }).print();

        env.execute();
    }

    // 自定义MapFunction的实现类
    public static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String, Long>>{
        @Override
        public Tuple2<String, Long> map(Event value) throws Exception {
            return Tuple2.of(value.user, 1L);
        }
    }
}

max maxby 区别

DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 5000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Mary", "./cart", 3000L),
                new Event("ss", "./fav", 4000L),
                new Event("Mary", "./fav", 10000L)
        );
                
stream.keyBy(e -> e.user)
//                .maxBy("timestamp")
                .maxBy("timestamp")    // 指定字段名称
                .print("maxBy:");
        stream.keyBy(e -> e.user)
//                .("timestamp")
                .max("timestamp")    // 指定字段名称
                .print("max:");       
               
               
max:> Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:05.0}
maxBy:> Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:05.0}
max:> Event{user='Bob', url='./cart', timestamp=1970-01-01 08:00:02.0}
maxBy:> Event{user='Bob', url='./cart', timestamp=1970-01-01 08:00:02.0}
max:> Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:05.0}
max:> Event{user='ss', url='./fav', timestamp=1970-01-01 08:00:04.0}
maxBy:> Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:05.0}
max:> Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:10.0}
maxBy:> Event{user='ss', url='./fav', timestamp=1970-01-01 08:00:04.0}
maxBy:> Event{user='Mary', url='./fav', timestamp=1970-01-01 08:00:10.0}


max部分替换，maxby全部替换

2 窗口

1 简介

窗口[0-10）中有11,12,但是11,12并不在窗口[0-10）处理，而是在对应的窗口[10,20)处理

2 窗口的分类

按照驱动类型分类

（1）时间窗口 Time Window

（2）计数窗口 Count Window

按照窗口分配数据的规则分类

（1）滚动窗口 Tumbling Windows

（2）滑动窗口 Sliding Windows

（3）会话窗口 Session Windows

（4）全局窗口 Global Windows

3 使用

4 迟到数据的处理

窗口中的迟到数据默认会被丢弃，这导致计算结果不够准确

1 设置水位线延迟时间

assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.timestamp;
            }
        }));

2 允许窗口处理迟到数据