技术标签: flink hive # Iceberg专题 sql
(1)下载jar包
iceberg-flink-runtime-1.14-1.0.0jar
flink-sql-connector-hive-2.3.6_2.12-1.11.2.jar
https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/table/hive/overview/
(2)启动FlinkSQL
①StandLone模式启动
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
./bin/sql-client.sh embedded -j <flink-runtime-directory>/iceberg-flink-runtime-xxx.jar shell
②Flink On Yarn模式启动
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
# 第一步 - 在Yarn集群上生成一个Standlone集群
./yarn-session.sh -s 2 -jm 2048 -tm 2048 -nm flinksql1 -d
# 第二步 - 指定yarn-session模式启动sql-client
./sql-client.sh embedded -s yarn-session -j ../lib/iceberg-flink-runtime-1.14-0.14.1.jar shell
CREATE CATALOG <catalog_name> WITH (
'type'='iceberg',
`<config_key>`=`<config_value>`
);
type
: 必须的iceberg
。(必需的)catalog-type
:hive
或hadoop
用于内置目录,或未设置用于使用 catalog-impl 的自定义目录实现。(可选的)catalog-impl
:自定义目录实现的完全限定类名。如果未设置,则必须catalog-type
设置。(可选的)property-version
: 描述属性版本的版本号。如果属性格式发生变化,此属性可用于向后兼容。当前的属性版本是1
. (可选的)cache-enabled
: 是否启用目录缓存,默认值为true
(1)创建Catalog
CREATE CATALOG hive_iceberg WITH (
'type'='iceberg',
'catalog-type'='hive',
'uri'='thrift://leidi01:9083',
'clients'='5',
'property-version'='1',
'hive-conf-dir'='/usr/hdp/3.1.0.0-78/hive/conf'
);
show catalogs;
uri
: Hive 元存储的 thrift URI。(必需的)clients
:Hive Metastore 客户端池大小,默认值为 2。(可选)warehouse
:Hive 仓库位置,如果既不设置hive-conf-dir
指定包含hive-site.xml
配置文件的位置也不添加正确hive-site.xml
的类路径,用户应指定此路径。hive-conf-dir``hive-site.xml
:包含将用于提供自定义 Hive 配置值的配置文件的目录的路径。如果同时设置和创建冰山目录时,hive.metastore.warehouse.dir
from /hive-site.xml
(或来自类路径的 hive 配置文件)的值将被该值覆盖。warehouse``hive-conf-dir``warehouse
(2)多客户端共享验证
(1)创建Catalog
CREATE CATALOG hadoop_catalog WITH (
'type'='iceberg',
'catalog-type'='hadoop',
'warehouse'='hdfs://leidi01:8020/warehouse/iceberg_catalog',
'property-version'='1'
);
warehouse
:HDFS目录,存放元数据文件和数据文件。(必需的)(1)创建自定义目录
catalog-impl
属性来加载自定义的 Iceberg实现REATE CATALOG my_catalog WITH (
'type'='iceberg',
'catalog-impl'='com.my.custom.CatalogImpl',
'my-additional-catalog-config'='my-value'
);
(2)通过SQL文件创建目录
-- define available catalogs
CREATE CATALOG hive_catalog WITH (
'type'='iceberg',
'catalog-type'='hive',
'uri'='thrift://leidi01:9083',
'warehouse'='hdfs://leidi01:8020/user/flink/iceberg'
);
USE CATALOG hive_catalog;
sql-client-defaults.yaml
在 flink 1.14 中删除了该文件,需要初始化才能有文件。(1)建库建表
use catalog iceberg;
CREATE DATABASE iceberg_db;
USE iceberg_db;
CREATE TABLE iceberg.iceberg_db.iceberg_001 (
id BIGINT COMMENT 'unique id',
data STRING
) WITH ('connector'='iceberg','write.format.default'='ORC');
(2)创建分区table
CREATE TABLE iceberg.iceberg_db.iceberg_003 (
id BIGINT COMMENT 'unique id',
data STRING
) PARTITIONED BY (data);
(3)更改table
--1、CREATE TABLE LIKE
CREATE TABLE `hive_catalog`.`default`.`sample` (
id BIGINT COMMENT 'unique id',
data STRING
);
CREATE TABLE `hive_catalog`.`default`.`sample_like` LIKE `hive_catalog`.`default`.`sample`
--2、alter table
ALTER TABLE `hive_catalog`.`default`.`sample` SET ('write.format.default'='avro')
--3、ALTER TABLE .. RENAME TO
ALTER TABLE `hive_catalog`.`default`.`sample` RENAME TO `hive_catalog`.`default`.`new_sample`;
--4、DROP TABLE
DROP TABLE `hive_catalog`.`default`.`sample`;
(1)插入数据
INSERT INTO `iceberg`.`iceberg_db`.`iceberg_001` VALUES (1, 'a');
--分区表插入语句
INSERT INTO `iceberg`.`iceberg_db`.`iceberg_001`() values(2,'b')
INSERT OVERWRITE sample VALUES (1, 'a');
(2)查询数据
-- Execute the flink job in streaming mode for current session context
SET execution.runtime-mode = streaming;
-- Execute the flink job in batch mode for current session context
SET execution.runtime-mode = batch;
Ⅰ、批量读取:通过提交 flink批处理作业来检查 iceberg 表中的所有行
SET execution.runtime-mode = batch;
SELECT * FROM sample;
Ⅱ、流式读取:支持处理从历史快照 id 开始的 flink 流作业中的增量数据
-- Submit the flink job in streaming mode for current session.
SET execution.runtime-mode = streaming;
-- Enable this switch because streaming read SQL will provide few job options in flink SQL hint options.
SET table.dynamic-table-options.enabled=true;
-- Read all the records from the iceberg current snapshot, and then read incremental data starting from that snapshot.
SELECT * FROM sample /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/ ;
-- Read all incremental data starting from the snapshot-id '3821550127947089987' (records from this snapshot will be excluded).
SELECT * FROM sample /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-snapshot-id'='3821550127947089987')*/ ;
(3)更新数据
前提:启动更新模式
模式一:启用UPSERT
模式作为表级属性write.upsert.enabled
CREATE TABLE `hive_catalog`.`default`.`sample` (
`id` INT UNIQUE COMMENT 'unique id',
`data` STRING NOT NULL,
PRIMARY KEY(`id`) NOT ENFORCED
) with ('format-version'='2', 'write.upsert.enabled'='true');
write options
中使用启用UPSERT
模式upsert-enabled
提供了比表级配置更大的灵活性。INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
...
(1)创建Catalog
CREATE CATALOG hadoop_catalog WITH (
'type'='iceberg',
'catalog-type'='hadoop',
'warehouse'='hdfs://leidi01:8020/warehouse/iceberg_catalog',
'property-version'='1'
);
一个catalog + 一个默认的default数据库
(2)查看HDFS结构目录
(1)建库建表
create database hadoop_test;
use hadoop_test;
CREATE TABLE `hadoopdemo` (
> id BIGINT COMMENT 'unique id',
> data STRING
> );
(2)查看对应HDFS目录
catalog为一级目录、数据库为二级目录、表为三级目录
,建Catalog、建库、建表时没有flink任务生成。(1)插入数据
INSERT INTO `iceberg`.`iceberg_db`.`iceberg_001` VALUES (1, 'a');
(2)HDFS目录
①data目录文件结构
②metadata目录文件结构
文件名称 | 文件描述 | 备注 |
---|---|---|
version[number].metadata.json | 存储每个版本的数据更改项 | |
snap-[snapshotID]-[attemptID]-[commitUUID].avro | 存储快照snapshot文件; | |
[commitUUID]-[attemptID]-[manifestCount].avro | 清单文件,每次更新操作都会产生清单文件 | |
version-hint.text |
Hive metastore 中的表可以表示加载 Iceberg 表的三种不同方式,具体取决于表的iceberg.catalog
属性:
如果在Hive中创建Iceberg格式表时不指定Iceberg.catalog属性,将使用HiveCatalog
与 Hive 环境中配置的 Metastore 相对应的表加载该表iceberg.catalog
,那么数据存储在对应的Hive Warehouse路径下。
-- 1、在Hive中创建Iceberg格式表
create table test_iceberg_tbl1(
id int,
name string,
age int)
partitioned by (dt string)
stored by 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';
-- 2、在Hive中加载如下两个包,在向Hive中插入数据时执行MR程序时需要使用到
add jar /usr/hdp/3.1.0.0-78/hive/lib/iceberg-hive-runtime-0.14.1.jar
add jar /usr/hdp/3.1.0.0-78/hive/lib/libfb303-0.9.3.jar
-- 3、向表中插入数据
insert into test_iceberg_tbl1 values(1,"sz",18,"beijing")
-- 4、查询表中数据
select * from test_iceberg_tbl1
iceberg.catalog
如果设置为Hive目录名称,将使用自定义目录加载该表 在Hive中创建Iceberg格式表时,如果指定了iceberg.catalog属性值,那么数据存储在指定的catalog名称对应配置的目录下。
-- 1、注册一个HiveCatalog叫another_hive
set iceberg.catalog.another_hive.type=hive;
SET iceberg.catalog.another_hive.uri=thrift://10.201.0.202:49153;
SET iceberg.catalog.another_hive.warehouse=s3a://faas-ethan/warehouse/;
SET hive.vectorized.execution.enabled=false;
-- 2、在Hive中创建iceberg格式表
create table test_iceberg_tbl2(
id int,
name string,
age int
)
partitioned by (dt string)
stored by 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
location 's3a://faas-ethan/warehouse/default/sample_hive_table_1'
tblproperties ('iceberg.catalog'='another_hive');
-- 3、插入数据,并查询
hive> insert into test_iceberg_tbl2 values (2,"ls",20,"20211212");
hive> select * from test_iceberg_tbl2;
在Hive中创建Iceberg表,会在两边HMS分别存储一份元数据,只有这样,远端HMS中的Iceberg表才对本地HMS可见,所以必须保证远端HMS存在对应的数据库。
问题:如果只有远端HMS的Iceberg表,如何在本地HMS访问?
解决方案:通过如下创建external外表的形式在本地HMS生成元数据。
CREATE EXTERNAL TABLE default.sample_hive_table_1(
id bigint, name string
)
PARTITIONED BY(
dept string
)
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
location 's3a://faas-ethan/warehouse/default/sample_hive_table_1'
TBLPROPERTIES ('iceberg.catalog'='another_hive');
select * from default.sample_local_hive_table_1,sample_hive_table_1;
iceberg.catalog
如果设置为location_based_table,则可以使用表的根位置直接加载表location_based_table
如果HDFS中已经存在iceberg格式表,我们可以通过在Hive中创建Icerberg格式表指定对应的location路径映射数据。
CREATE TABLE test_iceberg_tbl4 (
id int,
name string,
age int,
dt string
)STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
LOCATION 'hdfs://leidi01:8020/flinkiceberg/iceberg_db/flink_iceberg_tbl2'
TBLPROPERTIES ('iceberg.catalog'='location_based_table');
--指定的location路径下必须是iceberg格式表数据,并且需要有元数据目录才可以。不能将其他数据映射到Hive iceberg格式表。
由于Hive建表语句分区语法Partitioned by
的限制,如果使用Hive创建Iceberg格式表,目前只能按照Hive语法来写,底层转换成Iceberg标识分区,这种情况下不能使用Iceberge的分区转换,例如:days(timestamp),如果想要使用Iceberg格式表的分区转换标识分区,需要使用Spark或者Flink引擎创建表。
SET iceberg.catalog.hadoop_cat.type=hadoop;
SET iceberg.catalog.hadoop_cat.warehouse=s3a://faas-ethan/warehouse;
CREATE TABLE default.sample_hadoop_table_1(
id bigint, name string
) PARTITIONED BY (
dept string
)
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
LOCATION 's3a://faas-ethan/warehouse/default/sample_hadoop_table_1'
TBLPROPERTIES ('iceberg.catalog'='hadoop_cat');
Hadoop Catalog
相比Hive Catalog
建立的表相比,少了metadata_location
属性,同时元数据文件多了 version-hint.text
。
文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下:http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态:[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态
文章浏览阅读1.3w次,点赞45次,收藏99次。我个人用的是anaconda3的一个python集成环境,自带jupyter notebook,但在我打开jupyter notebook界面后,却找不到对应的虚拟环境,原来是jupyter notebook只是通用于下载anaconda时自带的环境,其他环境要想使用必须手动下载一些库:1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境
文章浏览阅读5.2k次,点赞19次,收藏28次。选择scoop纯属意外,也是无奈,因为电脑用户被锁了管理员权限,所有exe安装程序都无法安装,只可以用绿色软件,最后被我发现scoop,省去了到处下载XXX绿色版的烦恼,当然scoop里需要管理员权限的软件也跟我无缘了(譬如everything)。推荐添加dorado这个bucket镜像,里面很多中文软件,但是部分国外的软件下载地址在github,可能无法下载。以上两个是官方bucket的国内镜像,所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn
文章浏览阅读4.5k次,点赞2次,收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了,例如:这里的:sty..._vue el-color-picker
文章浏览阅读640次。基于芯片日益增长的问题,所以内核开发者们引入了新的方法,就是在内核中只保留函数,而数据则不包含,由用户(应用程序员)自己把数据按照规定的格式编写,并放在约定的地方,为了不占用过多的内存,还要求数据以根精简的方式编写。boot启动时,传参给内核,告诉内核设备树文件和kernel的位置,内核启动时根据地址去找到设备树文件,再利用专用的编译器去反编译dtb文件,将dtb还原成数据结构,以供驱动的函数去调用。firmware是三星的一个固件的设备信息,因为找不到固件,所以内核启动不成功。_exynos 4412 刷机
文章浏览阅读2w次,点赞24次,收藏42次。Linux系统配置jdkLinux学习教程,Linux入门教程(超详细)_linux配置jdk
文章浏览阅读3.3k次,点赞5次,收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图:_matlab微米怎么输入
文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件,可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中,对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数,而关闭则使用fclose函数。在C语言中,可以使用fread和fwrite函数进行二进制读写。 Biaoge 于2024-03-09 23:51发布 阅读量:7 ️文章类型:【 C语言程序设计 】在C语言中,用于打开文件的函数是____,用于关闭文件的函数是____。
文章浏览阅读3.4k次,点赞2次,收藏13次。跟随鼠标移动的粒子以grid(SOP)为partical(SOP)的资源模板,调整后连接【Geo组合+point spirit(MAT)】,在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in(CHOP)鼠标位置到metaball的坐标,实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动
文章浏览阅读178次。项目运行环境配置:Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX(Webstorm也行)+ Eclispe(IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持)。项目技术:Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成,B/S模式 + Maven管理等等。环境需要1.运行环境:最好是java jdk 1.8,我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计
文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多,会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析,后面会慢慢更新。另外,博客只作为自己学习记录的一种方式,对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析 时序图
文章浏览阅读2.4k次,点赞41次,收藏13次。java 数据结构与算法 ——快速排序法_快速排序法