大数据架构发展历史2.

format从如上iceberg的数据结构可以知道iceberg在数据查询时1.查找文件的时间复杂度至少是O(1)2.

iceberg

数据仓库加载各个数据源到HIVE、HBASE等数据湖数据入湖-再建仓多中数据源、或ETL湖仓一体数据入湖、湖上建仓。

离线实时数据使用同一批数据。

统一口径离线、实时、ad-hoc、机器学习都可以使用同一个数据源数据治理简单。

Lambda架构与kappa架构

复杂性分为速度层批层流批不同的技术维护两套不同的代码库、工具维护成本很高流、批分离处理相同数据出现不一致的结果延迟流等批增加延迟。

CDC可以解决

数据回溯能力弱面对更复杂的数据分析时要将DWD和DWS层的数据写入到ClickHouse、ES、MySQL或者是Hive里做进一步分析这无疑带来了链路的复杂性。

OLAP分析能力弱Kafka是一个顺序存储的系统顺序存储系统是没有办法直接在其上进行OLAP分析的例如谓词下推这类的优化策略在顺序存储平台Kafka上实现是比较困难的事情。

数据时序性受到挑战Kappa架构是严重依赖于消息队列的我们知道消息队列本身的准确性严格依赖它上游数据的顺序但是消息队列的数据分层越多发生乱序的可能性越大。

这里可以将kafka改为将starrocks或doris作为实时数仓的存储层以及olap分析层。

既能够支持数据高效的回溯能力支持数据的更新ACID又能够实现数据的批流读写并且还能够实现分钟级到秒级的数据接入

有没有这样一个架构

既能够满足实时性的需求又能够满足离线计算的要求而且还能够减轻开发运维的成本解决通过消息队列方式构建的Kappa架构中遇到的痛点

流批数据处理的统一与能力数据入湖后支持对数据的修正、数据质量管理的能力。

数据的一致性与正确性ACID事务的能力元数据的可拓展性。

计算引擎与存储引擎的解耦这样数据湖中间件可以在多个地方应用即在不同计算引擎spark、flink、trino、hive、starrocks…、存储引擎hdfs、s3上应用。

该项目最初是由Netflix公司开发的目的是解决他们使用巨大的PB级表的长期问题。

它于2018年作为Apache孵化器项目开源并于2020年5月19日从孵化器中毕业。

表格式Table

Format是对元数据以及数据文件的一种组织方式处于计算框架FlinkSpark…之下数据文件之上。

我们先回到Netflix

不能在原表之上直接修改只能新建一个按小时分区的表再把数据Insert到新的小时分区表。

因为分区字段修改导致需要修改原表上层的应用的sql即使通过Rename的命令把新表的名字改为原表。

所以数据的组织方式表格式是许多数据基础设施面临挫折和问题的共同原因。

Apache

Iceberg设计的一个关键考虑是解决各种数据一致性和性能问题这些问题是Hive在使用大数据时所面临的问题。

]

state存储在两个地方分区存储在hive元数据、文件存储在文件系统。

bucketing分桶是由hive的hash实现效率不高吗非

ACID

Iceberg表真实存储数据的文件一般是在表的数据存储下iceberg支持三种格式parquet、avro、orc的文件存储。

表快照

快照代表一张表在某个时刻下表的状态。

每个快照里面会列出表在某个时刻的所有

data

list是一个元数据文件它列出构建表快照Snapshot的清单Manifest

manifest

files的分区范围增加了几个数文件、删除了几个数据文件等信息

清单文件

数据文件的状态、文件路径、分区信息、列级别的统计信息比如每列的最大最小值、空值数等、文件的大小以及文件里面数据行数等信息。

从如上iceberg的数据结构可以知道iceberg在数据查询时1.查找文件的时间复杂度至少是O(1)2.

iceberg

序号特性说明1统一存储统一性数据都统一存储到hdfs、s3中。

-

数据湖中可以存储结构化、半结构化、非结构化数据我们可以通过iceberg来摄取这些数据。

但要注意数据湖存储例如图片等非结构化数据并不是强项。

2插件化灵活性Iceberg不和特定的数据存储、计算引擎绑定。

常见数据存储HDFS、S3…计算引擎Flink、Spark…都可以接入Iceberg。

3模式演化演化能力支持table、schema、Partition的添加、删除、更新或重命名简化表修改成本。

4隐藏分区分区信息并不需要人工维护会自动计算。

由于Iceberg的分区信息和表数据存储目录是独立的使得Iceberg的表分区可以被修改,而且不涉及到数据迁移。

5Time

Travel镜像数据查询允许用户通过将表重置为之前某一时刻的状态来快速纠正问题。

6乐观锁的并发支持提供了多个程序并发写入的能力并且保证数据线性一致。

7支持事务upsert与读写分离-

提供事务ACID的机制使其具备了upsert的能力并且使得边写边读成为可能从而数据可以更快的被下游组件消费。

-

通过事务保证了下游组件只能消费已commit的数据而不会读到部分甚至未提交的数据。

8文件级数据剪裁文件级谓词下推-

Iceberg的元数据里面提供了每个数据文件的一些统计信息比如最大值最小值Count计数等等。

-

查询SQL的过滤条件除了常规的分区列过滤甚至可以下推到文件级别大大加快了查询效率。

iceberg不支持自动文件合并历史数据也需要自己手动清洗文件格式支持的最多parquet、avro、orc存储引擎支持hdfs、S3不支持索引

iceberg实战

iceberg-flink-runtime-1.16-1.5.2.jar

flink/lib

https://iceberg.apache.org/docs/latest/java-api-quickstart/

2.1.

{hadoopCatalog.createNamespace(Namespace.of(namespaceName));System.out.println(创建Namespace成功);}Overridepublic

void

{hadoopCatalog.dropNamespace(Namespace.of(namespaceName));System.out.println(删除Namespace成功);}Overridepublic

Namespace

(hadoopCatalog.namespaceExists(Namespace.of(namespaceName)))

{//

hadoopCatalog.listNamespaces(Namespace.of(namespaceName)).get(0);}return

Namespace.empty();}Overridepublic

ListNamespace

hadoopCatalog.listNamespaces();}Overridepublic

Table

从其他模式格式如Spark、Avro和Parquet进行转换时将自动分配新的ID

*/Schema

Schema(Types.NestedField.required(1,

level,

Types.StringType.get()),Types.NestedField.required(2,

event_time,

Types.TimestampType.withZone()),Types.NestedField.required(3,

message,

Types.StringType.get()),Types.NestedField.optional(4,

call_stack,Types.ListType.ofRequired(5,

分区规范描述了Iceberg如何将记录分组成数据文件。

分区规范是使用构建器为表的模式创建的。

**

p以下是按照日志事件的时间戳的小时和日志级别进行分区*/PartitionSpec

partition

PartitionSpec.builderFor(schema).hour(event_time).identity(level).build();//

table

hadoopCatalog.createTable(spaceAndTableName,

schema,

partition);System.out.println(创建表

table

{hadoopCatalog.dropTable(TableIdentifier.of(namespaceName,

Table

hadoopCatalog.listTables(Namespace.of(namespaceName));}Overridepublic

Configuration

Configuration();configuration.set(fs.defaultFS,

configuration.addResource(new//

Path(/Users/lianggao/MyWorkSpace/001-360/001project-360/datalake-metadata-api/datalake-metadata-iceberg/src/main/resources/core-site.xml));//

configuration.addResource(new//

Path(/Users/lianggao/MyWorkSpace/001-360/001project-360/datalake-metadata-api/datalake-metadata-iceberg/src/main/resources/hdfs-site.xml));//

configuration.addResource(new//

Path(/usr/hdp/current/hive-client/conf/hdfs-site.xml));configuration.set(fs.hdfs.impl,

org.apache.hadoop.hdfs.DistributedFileSystem);//

configuration.setBoolean(fs.hdfs.impl.disable.cache,

true);//

configuration.set(hadoop.job.ugi,

logsget);//

UserGroupInformation.setConfiguration(configuration);//

try

subject.getPrincipals().add(new

UserGroupInformation.loginUserFromSubject(null);//

catch

standalone集群去提交iceberg表相关操作如下创建catalog我们看到创建的catalog持久化到了s3存储中。

表操作

warehousehdfs://iceberg1v.middle.bjmd.qihoo.net:9000/warehouse/iceberg-hadoop,

property-version1

STRING,order_amount_daily_category_city

decimal(20,2)

https://iceberg.apache.org/docs/nightly/flink/

表历史

spotify$metadata_log_entries;--

snapshots

下运行的交互慢费资源所以不推荐使用flink对catalog进行管理而是使用native

自己搭建的集群与现有系统部环境暂不统一使用系统部的hadoop作为数据湖的存储

Flink

warehousehdfs://iceberg1v.middle.bjmd.qihoo.net:9000/warehouse/iceberg-hadoop,

property-version1

viewfs://iceberg1v.middle.bjmd.qihoo.net:9000/Flink

SQL

warehousehdfs://namenode.dfs.shbt.qihoo.net:9000/home/logsget/warehouse/iceberg-hadoop,

property-version1

order_amount_daily_category_city

decimal(20,2));

org.apache.hadoop.hdfs.DFSClient

Cannot

/home/logsget/warehouse/iceberg-hadoop/default/sample/metadata/version-hint.text:

such

warehousehdfs://namenode.dfs.shbt.qihoo.net:9000/home/logsget/warehouse/iceberg-hadoop,

property-version1

order_amount_daily_category_city

decimal(20,2)

dp_test,sink.buffer-flush.interval-ms

5000,fe-nodes

db01.doris.shyc2.qihoo.net:8030,table-name

ads_product_citycategoryamount_di,url

jdbc:mysql://10.192.197.134:9030/dp_test?useUnicodetruecharacterEncodingutf-8useSSLfalseconnectTimeout3000useUnicodetruecharacterEncodingutf8useSSLfalserewriteBatchedStatementstrueserverTimezoneAsia/ShanghaisessionVariablesquery_timeout86400,username

dfs_shbt_logsget

aaa_b;/data01/chunjun-master-dev/bin/run-ri-test.sh

/data01/chunjun-master-dev/conf/ice-w.sql

offline

warehousehdfs://namenode.dfs.shbt.qihoo.net:9000/home/logsget/warehouse/iceberg-hadoop,

property-version1

order_amount_daily_category_city

decimal(20,2)

;/data01/chunjun-master-dev/bin/run-ri-test.sh

/data01/chunjun-master-dev/conf/ice-w.sql

\offline

\radar_1_2187_9270_4058632_test

local

虽然iceberg当初是为了解决hive表格式的问题但实际上iceberg的种种能力使得他配得上作为数据湖中间件这里再回顾下iceberg的能力

流批数据处理的统一与能力数据入湖后支持对数据的修正、数据质量管理的能力。

数据的一致性与正确性ACID事务的能力元数据的可拓展性。

计算引擎与存储引擎的解耦这样数据湖中间件可以在多个地方应用即在不同计算引擎spark、flink、trino、hive、starrocks…、存储引擎hdfs、s3上应用。

我们可以借助iceberg搭建存储统一、计算口径统一的数据湖仓。

使用iceberg

api进行元数据管理使用flink进行数据入湖湖仓建设使用flink、spark、trino、hive等进行数据分析。

flink

iceberg数据入湖的事务能力验证iceberg修改表结构对任务的影响iceberg批数据分批写完下游数据立马能消费的验证以及相关原理iceberg数据合并的逻辑验证。

iceberg处理非结构数据与半结构数据的实践