setParallelism

reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                        // 将累加器更新为当前最大的pv统计值，然后向下游发送累加器的值
                        return value1.f1 > value2.f1 ? value1 : value2;
                    }
                }).setParallelism(5)
                .print();
                
(Mary,1)
(Bob,1)
(Mary,2)
(Bob,2)
(Mary,3)
(Bob,3)
(Mary,4)
(Bob,4)


keyBy(r -> true) // 为每一条数据分配同一个key，将聚合结果发送到一条流中去
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                        // 将累加器更新为当前最大的pv统计值，然后向下游发送累加器的值
                        return value1.f1 > value2.f1 ? value1 : value2;
                    }
                })
                .print().setParallelism(5);
                
                
2> (Bob,2)
1> (Mary,2)
1> (Bob,4)
3> (Mary,3)
5> (Bob,1)
5> (Mary,4)
4> (Mary,1)
4> (Bob,3)

2Execution Environment Level（执行环境层次）

3Client Level（客户端层次）

4System Level（系统层次）

优先级1>2>3>4

大数据基础组件 flink 使用

并行度设置

2022-05-09 9c34b49acf8c16862d41bb93619059c3 99+ a minute 0.2 k

物理分区

https://www.cnblogs.com/wdh01/p/16038278.html

首先和逻辑分区区别开，逻辑分区包括keyBy等算子

逻辑分区只不过将数据按照key分组，哪个key分到哪个task，系统自动控制，万一分配不均，会发生数据倾斜

物理分区就是按一定逻辑将数据分配到不同Task，可以缓解数据倾斜

source（1）-》不同物理分区方式（3）-》slot

分类

1 随机分区 random

2 轮询分区round-robin

3 重缩放分区 rescale

4 分局分区 global

5 自定义 custom

6 广播

不完全算物理分区方式

大数据基础组件 flink 使用

物理分区

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-05-01 c65350409e5b19c4bbac3a2872f1b2e6 99+ 3 m 0.5 k