[大数据]Hudi编译集成

news2025/2/27 17:12:19

1. Hudi概述

1.1 Hudi简介 What is Apache Hudi

Apache Hudi is the next generation streaming data lake platform. Apache Hudi brings core warehouse and database functionality directly to a data lake. Hudi provides tables, transactions, efficient upserts/deletes, advanced indexes, streaming ingestion services, data clustering/compaction optimizations, and concurrency all while keeping your data in open source file formats.

Apache Hudi 是下一代流式数据湖平台。Apache Hudi 将核心的数据仓库和数据库功能直接引入数据湖。Hudi 提供了表、事务、高效的更新/删除、先进的索引、流式摄取服务、数据聚类/压缩优化以及并发处理,同时保持数据在开源文件格式中。

行式存储: .avro

列式存储: .parquet

字节跳动基于Apache Hudi构建EB级数据湖实践

Not only is Apache Hudi great for streaming workloads, but it also allows you to create efficient incremental batch pipelines. Read the docs for more use case descriptions and check out who's using Hudi, to see how some of the largest data lakes in the world including Uber, Amazon, ByteDance, Robinhood and more are transforming their production data lakes with Hudi.

Apache Hudi 不仅适用于流式工作负载,还允许您创建高效的增量批处理管道。请阅读文档以获取更多用例描述,并查看谁在使用 Hudi,了解一些全球最大的数 据湖如何利用 Hudi 转型他们的生产数据湖,包括 Uber、亚马逊、字节跳动、Robinhood 等公司。 

Apache Hudi can easily be used on any cloud storage platform. Hudi’s advanced performance optimizations, make analytical workloads faster with any of the popular query engines including, Apache Spark, Flink, Presto, Trino, Hive, etc.

Apache Hudi 可以轻松地在任何云存储平台上使用。Hudi 的高级性能优化可以加快与任何流行查询引擎(包括 Apache Spark、Flink、Presto、Trino、Hive 等)一起进行的分析工作负载。

2. 编译Hudi

2.1 编译环境准备

相关组件版本如下:

Hadoop3.3.1
Hive3.1.3
Flink1.13.6,scala-2.12
Spark3.3.1,scala-2.12

1)安装Maven

(1)上传apache-maven-3.6.1-bin.tar.gz到/opt/software目录,并解压更名

tar -zxvf apache-maven-3.6.1-bin.tar.gz -C /opt/module/

mv apache-maven-3.6.1 maven-3.6.1

(2)添加环境变量到/etc/profile中

sudo vim /etc/profile.d/my_env.sh

#MAVEN_HOME

export MAVEN_HOME=/opt/module/maven-3.6.1

export PATH=$PATH:$MAVEN_HOME/bin

(3)测试安装结果

source /etc/profile

mvn -v

2)修改为阿里镜像

(1)修改setting.xml,指定为阿里仓库地址

vim /opt/module/maven-3.6.1/conf/settings.xml

<!-- 添加阿里云镜像-->
<mirror>
    <id>nexus-aliyun</id>
    <mirrorOf>central</mirrorOf>
    <name>Nexus aliyun</name>
    <url>http://maven.aliyun.com/nexus/content/groups/public</url>
</mirror>

2.2 下载源码包

https://github.com/apache/hudi/blob/release-0.12.3/README.md

wget https://dlcdn.apache.org/hudi/0.12.3/hudi-0.12.3.src.tgz

1)修改pom.xml,添加以下仓库地址:

<repository>
    <id>nexus-aliyun</id>
    <name>nexus-aliyun</name>
    <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
    <releases>
        <enabled>true</enabled>
    </releases>
    <snapshots>
        <enabled>false</enabled>
    </snapshots>
</repository>

 2)修改依赖的组件版本

<hadoop.version>3.3.1</hadoop.version>
<hive.version>3.1.3</hive.version>

 

2.3 编译命令

mvn clean package -DskipTests -Dspark3.3 -Dflink1.13 -Dscala-2.12 -Dhadoop.version=3.3.1 -Pflink-bundle-shade-hive3

直接编译报错:

hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java

原因:hadoop2 和hadoop3兼容性问题

hadoop2 的API:

修改为hadoop3的API:

其他地方都不修改:

 

 

 

 

 2.4 解决Spark模块依赖冲突

java.lang.AbstractMethodError: org.apache.hudi.org.apache.jetty.server.AllowedResourceAliasChecker$AllowedResourceAliasCheckListener.lifeCycleFailure(Lorg/apache/hudi/org/apache/jetty/util/component/LifeCycle;Ljava/lang/Throwable;)V

 修改了Hive版本为3.1.3,其携带的jetty是0.9.3,hudi本身用的0.9.4,存在依赖冲突。
1)修改hudi-spark-bundle的pom文件,排除低版本jetty,添加hudi指定版本的jetty:
vim /opt/software/hudi-0.12.3/packaging/hudi-spark-bundle/pom.xml

    <!-- Hive -->
    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-service</artifactId>
      <version>${hive.version}</version>
      <scope>${spark.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <artifactId>servlet-api</artifactId>
          <groupId>javax.servlet</groupId>
        </exclusion>
        <exclusion>
          <artifactId>guava</artifactId>
          <groupId>com.google.guava</groupId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.pentaho</groupId>
          <artifactId>*</artifactId>
        </exclusion>

      </exclusions>
    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-service-rpc</artifactId>
      <version>${hive.version}</version>
      <scope>${spark.bundle.hive.scope}</scope>
    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-jdbc</artifactId>
      <version>${hive.version}</version>
      <scope>${spark.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>javax.servlet</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>javax.servlet.jsp</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>

    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-metastore</artifactId>
      <version>${hive.version}</version>
      <scope>${spark.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>javax.servlet</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.datanucleus</groupId>
          <artifactId>datanucleus-core</artifactId>
        </exclusion>
        <exclusion>
          <groupId>javax.servlet.jsp</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <artifactId>guava</artifactId>
          <groupId>com.google.guava</groupId>
        </exclusion>
      </exclusions>

    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-common</artifactId>
      <version>${hive.version}</version>
      <scope>${spark.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>org.eclipse.jetty.orbit</groupId>
          <artifactId>javax.servlet</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>

    </dependency>

    <!-- 增加hudi配置版本的jetty -->
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-server</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-util</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-webapp</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-http</artifactId>
      <version>${jetty.version}</version>
    </dependency>

2)修改hudi-utilities-bundle的pom文件,排除低版本jetty,添加hudi指定版本的jetty:
vim /opt/software/hudi-0.12.3/packaging/hudi-utilities-bundle/pom.xml
修改如下(红色部分)

    <!-- Hive -->
    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-service</artifactId>
      <version>${hive.version}</version>
      <scope>${utilities.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>org.apache.hbase</groupId>
          <artifactId>*</artifactId>
        </exclusion>
    <exclusion>
          <artifactId>servlet-api</artifactId>
          <groupId>javax.servlet</groupId>
        </exclusion>
        <exclusion>
          <artifactId>guava</artifactId>
          <groupId>com.google.guava</groupId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.pentaho</groupId>
          <artifactId>*</artifactId>
        </exclusion>

      </exclusions>
    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-service-rpc</artifactId>
      <version>${hive.version}</version>
      <scope>${utilities.bundle.hive.scope}</scope>
    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-jdbc</artifactId>
      <version>${hive.version}</version>
      <scope>${utilities.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>javax.servlet</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>javax.servlet.jsp</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>

    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-metastore</artifactId>
      <version>${hive.version}</version>
      <scope>${utilities.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>org.apache.hbase</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>javax.servlet</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.datanucleus</groupId>
          <artifactId>datanucleus-core</artifactId>
        </exclusion>
        <exclusion>
          <groupId>javax.servlet.jsp</groupId>
          <artifactId>*</artifactId>
        </exclusion>
        <exclusion>
          <artifactId>guava</artifactId>
          <groupId>com.google.guava</groupId>
        </exclusion>
      </exclusions>

    </dependency>

    <dependency>
      <groupId>${hive.groupid}</groupId>
      <artifactId>hive-common</artifactId>
      <version>${hive.version}</version>
      <scope>${utilities.bundle.hive.scope}</scope>
      <exclusions>
        <exclusion>
          <groupId>org.eclipse.jetty.orbit</groupId>
          <artifactId>javax.servlet</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.eclipse.jetty</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>

    </dependency>

    <dependency>
      <groupId>org.apache.htrace</groupId>
      <artifactId>htrace-core</artifactId>
      <version>${htrace.version}</version>
      <scope>compile</scope>
    </dependency>

    <!-- 增加hudi配置版本的jetty -->
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-server</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-util</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-webapp</artifactId>
      <version>${jetty.version}</version>
    </dependency>
    <dependency>
      <groupId>org.eclipse.jetty</groupId>
      <artifactId>jetty-http</artifactId>
      <version>${jetty.version}</version>
    </dependency>

再次编译:

mvn clean package -DskipTests -Dspark3.3 -Dflink1.13 -Dscala-2.12 -Dhadoop.version=3.3.1 -Pflink-bundle-shade-hive3 

 

3. 核心概念

3.1 基本概念

Apache Hudi核心概念一网打尽

3.1.1 时间轴(TimeLine)

At its core, Hudi maintains a timeline of all actions performed on the table at different instants of time that helps provide instantaneous views of the table, while also efficiently supporting retrieval of data in the order of arrival. A Hudi instant consists of the following components

Hudi的核心是维护表上在不同的即时时间(instants)执行的所有操作的时间轴(timeline),这有助于提供表的即时视图,同时还有效地支持按到达顺序检索数据。一个instant由以下三个部分组成:

在其核心,Hudi维护了一个时间轴,记录了表在不同时间点上执行的所有操作,这有助于提供即时的表视图,同时还能高效地支持按到达顺序检索数据。Hudi的一个时间点包括以下组件。

instantaneous:即时的,瞬时的

instant

n. 瞬息;顷刻;刹那;(某一)时刻

adj. 立即的;即刻的;速溶的;即食的;速食的;紧急的;紧迫的;迫切的;本月的;正在考虑中的

retrieval:[n.] 检索;恢复;取回;找回;恢复

  • Instant action : Type of action performed on the table 
  • Instant time : Instant time is typically a timestamp (e.g: 20190117010349), which monotonically increases in the order of action's begin time.
  • state : current state of the instant

即时行动:在表上执行的某种操作;
即时时间:通常是一个时间戳(例如:20190117010349),它按照动作开始时间的顺序单调递增。
状态:instant当前的状态。

monotonically:adv.单音调地;音调不变地;单调地

Hudi guarantees that the actions performed on the timeline are atomic & timeline consistent based on the instant time.

Hudi保证在时间轴上执行的操作基于即时时间是原子且时间一致的。

Key actions performed include

  • COMMITS - A commit denotes an atomic write of a batch of records into a table.
  • CLEANS - Background activity that gets rid of older versions of files in the table, that are no longer needed.
  • DELTA_COMMIT - A delta commit refers to an atomic write of a batch of records into a MergeOnRead type table, where some/all of the data could be just written to delta logs.
  • COMPACTION - Background activity to reconcile differential data structures within Hudi e.g: moving updates from row based log files to columnar [kəˈlʌmnə] formats. Internally, compaction manifests as a special commit on the timeline
  • ROLLBACK - Indicates that a commit/delta commit was unsuccessful & rolled back, removing any partial files produced during such a write
  • SAVEPOINT - Marks certain file groups as "saved", such that cleaner will not delete them. It helps restore the table to a point on the timeline, in case of disaster/data recovery scenarios.

主要的操作(Actions)包括:

  • 提交(COMMITS)- 提交表示将一批记录原子性地写入表中。
  • 清理(CLEANS)- 用于清除表中不再需要的旧文件版本的后台活动。
  • 增量提交(DELTA_COMMIT)- 增量提交指的是将一批记录原子性地写入MergeOnRead类型的表中,其中部分或全部数据可能刚刚写入增量日志。
  • 压缩(COMPACTION)- 后台活动,用于协调Hudi中的差异数据结构,例如将更新操作从基于行的日志文件合并到列式存储的数据文件中。在内部,Compaction表现为时间轴上的特殊提交。
  • 回滚(ROLLBACK)- 表示提交/增量提交失败并回滚,删除在此类写入期间生成的任何部分文件。
  • 保存点(SAVEPOINT)- 将某些文件组标记为“已保存”,以便清理器不会删除它们。它有助于在发生灾难需要数据恢复场景中将表恢复到时间轴上的某个点。

denote:vt. 表示;表明;代表;象征;意思是

reconcile [ˈrekənˌsaɪl]:vt.使接受;使顺从;使甘心于;使和解;使和好;使一致;使和谐;使(遭亵渎的教堂等)再次圣化。vi.和解;和好

Any given instant can be in one of the following states

  • REQUESTED - Denotes an action has been scheduled, but has not initiated
  • INFLIGHT - Denotes that the action is currently being performed
  • COMPLETED - Denotes completion of an action on the timeline

任何instant都处于以下状态之一:

REQUESTED——表示某个action已经调度,但尚未启动;

INFLIGHT——表示动作目前正在执行中;

COMPLETED——表示时间轴上的动作已完成。

两个时间概念:区分两个重要的时间概念:
    Arrival time: 数据到达 Hudi 的时间,commit time。
    Event time: record 中记录的时间。


 

上图中采用时间(小时)作为分区字段,从 10:00 开始陆续产生各种 commits,10:20 来了一条 9:00 的数据,根据event time该数据仍然可以落到 9:00 对应的分区,通过 timeline 直接消费 10:00 (commit time)之后的增量更新(只消费有新 commits 的 group),那么这条延迟的数据仍然可以被消费到。 

Example above shows upserts happenings between 10:00 and 10:20 on a Hudi table, roughly every 5 mins, leaving commit metadata on the Hudi timeline, along with other background cleaning/compactions. One key observation to make is that the commit time indicates the arrival time of the data (10:20AM), while the actual data organization reflects the actual time or event time, the data was intended for (hourly buckets from 07:00). These are two key concepts when reasoning about tradeoffs between latency and completeness of data.

上述示例显示,在Hudi表上发生了UPSERT操作,时间范围是10:00到10:20,大约每5分钟一次,并在Hudi时间轴上留下了提交元数据,以及其他后台清理/压缩。一个关键的观察点是,提交时间指示了数据到达的时间(上午10:20),而实际的数据组织反映了数据意图所在的实际时间或事件时间(从07:00开始的小时桶)。这是关于数据延迟和完整性的权衡时需要考虑的两个关键概念。

When there is late arriving data (data intended for 9:00 arriving >1 hr late at 10:20), we can see the upsert producing new data into even older time buckets/folders. With the help of the timeline, an incremental query attempting to get all new data that was committed successfully since 10:00 hours, is able to very efficiently consume only the changed files without say scanning all the time buckets > 07:00.

当有延迟到达的数据(预期9点到达的数据在10点20晚到达>1小时)时,我们可以看到upsert将新数据生成到更早的时段/文件夹中。在时间轴的帮助下,一个增量查询可以高效地获取自10点以来成功提交的所有新数据,只需要消费发生变化的文件即可,而不需要扫描所有大于7点的时间桶。

3.1.2 文件布局(File Layout)

Hudi将一个表映射为如下文件结构:

The following describes the general file layout structure for Apache Hudi

  • Hudi organizes data tables into a directory structure under a base path on a distributed file system
  • Tables are broken up into partitions
  • Within each partition, files are organized into file groups, uniquely identified by a file ID
  • Each file group contains several file slices
  • Each slice contains a base file (.parquet) produced at a certain commit/compaction instant time, along with set of log files (.log.*) that contain inserts/updates to the base file since the base file was produced.

Hudi adopts Multiversion Concurrency Control (MVCC), where compaction action merges logs and base files to produce new file slices and cleaning action gets rid of unused/older file slices to reclaim space on the file system.

以下描述了 Apache Hudi 的一般文件布局结构:

  • Hudi 将数据表组织成分布式文件系统上一个基础路径下的目录结构。  
  • 表被划分为多个分区。  
  • 在每个分区内,文件被组织成文件组,每个文件组由一个唯一的文件 ID 标识。  
  • 每个文件组包含多个文件切片。  
  • 每个切片包含一个基文件(.parquet),该文件是在某个提交/压缩瞬间生成的,以及一组日志文件(.log.*),这些日志文件包含自基文件生成以来对基文件的插入/更新。  

Hudi 采用多版本并发控制(MVCC),其中压缩操作将日志和基文件合并以生成新的文件切片,而清理操作则移除未使用的/旧的文件切片,以回收文件系统上的空间。

3.1.3 索引(Index)

3.1.4 表类型(Table Types)

3.1.5 查询类型(Query Types)

3.2 数据写

3.2.1 写操作

3.2.2 写流程(UPSERT)

3.2.3 写流程(INSERT)

3.2.4 写流程(INSERT OVERWRITE)

3.2.5 Key生成策略

3.2.6 删除策略

3.2.7 总结

3.3 数据读

3.3.1 Snapshot读

3.3.2 Incremental读

3.3.3 Streaming读

3.4 Compaction

4. 集成Spark

4.1 环境准备

4.2 spark-shell 方式

1)启动命令

# Spark 3.3
spark-shell \
--packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.12.3 \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

# Spark 3.2
spark-shell \
  --packages org.apache.hudi:hudi-spark3.2-bundle_2.12:0.12.3 \
  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
  --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

 # Spark 3.1
spark-shell \
  --packages org.apache.hudi:hudi-spark3.1-bundle_2.12:0.12.3 \
  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

# Spark 2.4
spark-shell \
  --packages org.apache.hudi:hudi-spark2.4-bundle_2.11:0.12.3 \
  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' 

Please note the following

  • For Spark 3.2 and above, the additional spark_catalog config is required: --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'
  • We have used hudi-spark-bundle built for scala 2.12 since the spark-avro module used can also depend on 2.12.

 2)设置表名,基本路径和数据生成器

// spark-shell
import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.common.model.HoodieRecord

val tableName = "hudi_trips_cow"
val basePath = "file:///tmp/hudi_trips_cow"
val dataGen = new DataGenerator

   存储选择对象存储:

// spark-shell
import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.common.model.HoodieRecord

val tableName = "hudi_trips_cow"
val basePath1 = "obs://bigdata-teach/tmp/hudi_trips_cow"
val dataGen = new DataGenerator

 The DataGenerator can generate sample inserts and updates based on the the sample trip schema here.

hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/QuickstartUtils.java

3)Create Table

// scala
// No separate create table command required in spark. First batch of write to a table will create the table if not exists.  

4)插入数据

新增数据,生成一些数据,将其加载到DataFrame中,然后将DataFrame写入Hudi表。 

// spark-shell
val inserts = convertToStringList(dataGen.generateInserts(10))
val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
df.write.format("hudi").
  options(getQuickstartWriteConfigs).
  option(PRECOMBINE_FIELD_OPT_KEY, "ts").
  option(RECORDKEY_FIELD_OPT_KEY, "uuid").
  option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
  option(TABLE_NAME, tableName).
  mode(Overwrite).
  save(basePath1)

Mode(overwrite)将覆盖重新创建表(如果已存在)。可以检查/tmp/hudi_trps_cow 路径下是否有数据生成。 

We provided a record key (uuid in schema), partition field (region/country/city) and combine logic (ts in schema) to ensure trip records are unique within each partition. For more info, refer to Modeling data stored in Hudi and for info on ways to ingest data into Hudi, refer to Writing Hudi Tables. Here we are using the default write operation : upsert. If you have a workload without updates, you can also issue insert or bulk_insert operations which could be faster. To know more, refer to Write operations .

我们提供了一个记录键(Schema中的uuid)、分区字段(地区/国家/城市)和组合逻辑(Schema中的ts),以确保每个分区内的行程记录是唯一的。有关更多信息,请参阅Hudi中存储的数据建模,以及有关将数据导入Hudi的方法,请参阅编写Hudi表。在这里,我们使用默认的写操作:upsert。如果您的工作负载没有更新操作,您还可以使用insert或bulk_insert操作,这可能更快。要了解更多信息,请参阅写操作。

4.3 spark-sql方式

# Spark 3.3
spark-sql --packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.12.3 \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

create table hudi_cow_pt_tbl (
id bigint,
name string,
ts bigint,
dt string,
hh string
) using hudi
tblproperties (
type = 'cow',
primaryKey = 'id',
preCombineField = 'ts'
)
partitioned by (dt, hh)
location 'obs://bigdata-test1233/tmp/tmp/hudi/hudi_cow_pt_tbl'; 

spark-sql (default)> show create table hudi_cow_nonpcf_tbl;
createtab_stmt
CREATE TABLE default.hudi_cow_nonpcf_tbl (
  _hoodie_commit_time STRING,
  _hoodie_commit_seqno STRING,
  _hoodie_record_key STRING,
  _hoodie_partition_path STRING,
  _hoodie_file_name STRING,
  uuid INT,
  name STRING,
  price DOUBLE)
USING hudi
LOCATION 'obs://bigdata-test1233/tmp/tmp/hudi/hudi_cow_pt_tbl'
TBLPROPERTIES (
  'primaryKey' = 'uuid',
  'type' = 'cow')
 

insert into hudi_cow_nonpcf_tbl select 1, 'a1', 20; 

 

 

4.4 IDEA集成

4.5 DeltaStreamer导入工具

4.6 并发控制

4.7 常规调优

5. 集成Flink

5.1 环境准备

5.2 sql-client方式

5.3 IDEA编码方式

5.4 类型映射

5.5 核心参数设置

5.6 内存优化

5.7 读取方式

5.8 限流

5.9 写入方式

5.10 写入模式

5.11 Bucket索引

5.12 Hudi Catalog

5.13 离线Compaction

5.14 离线Clustering

5.15 常见基础问题

5.16 核心原理分析

6. 集成Hive

6.1 集成步骤

6.2 Hive 同步

6.3 Flink使用HiveCatalog

6.4 创建Hive 外表

6.5 查询Hive外表

6.6 Hive sync tool

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2257983.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

Redis 数据结构(一)—字符串、哈希表、列表

Redis&#xff08;版本7.0&#xff09;的数据结构主要包括字符串&#xff08;String&#xff09;、哈希表&#xff08;Hash&#xff09;、列表&#xff08;List&#xff09;、集合&#xff08;Set&#xff09;、有序集合&#xff08;Sorted Set&#xff09;、超日志&#xff08…

Cisco Packet Tracer | Cisco Packet Tracer - VLAN 实验 - 交换机的 VLAN 划分

关注这个工具的其它相关笔记&#xff1a;Cisco Packet Tracer —— 使用教程合集-CSDN博客 0x01&#xff1a;VLAN 划分 - 单个交换机 0x0101&#xff1a;拓扑搭建流程 从软件底部拖出一台交换机&#xff08;笔者选择的型号是 2960 IOS15&#xff09;&#xff1a; 然后再拖出四…

元宇宙时代的社交平台:Facebook的愿景与实践

随着科技的不断进步&#xff0c;元宇宙&#xff08;Metaverse&#xff09;这一概念逐渐走进了人们的视野。作为全球最大的社交平台之一&#xff0c;Facebook&#xff08;现Meta&#xff09;在这场元宇宙革命中扮演着重要角色。Meta不仅在不断扩展其社交平台的边界&#xff0c;还…

Nignx部署Java服务测试使用的Spring Boot项目Demo

天行健&#xff0c;君子以自强不息&#xff1b;地势坤&#xff0c;君子以厚德载物。 每个人都有惰性&#xff0c;但不断学习是好好生活的根本&#xff0c;共勉&#xff01; 文章均为学习整理笔记&#xff0c;分享记录为主&#xff0c;如有错误请指正&#xff0c;共同学习进步。…

计算机网络ENSP课设--三层架构企业网络

本课程设计搭建一个小型互联网&#xff0c;并模拟Internet的典型Web服务过程。通过此次课程设计&#xff0c;可以进一步理解Internet的工作原理和协议过程&#xff0c;并提高综合知识的运用能力和分析能力。具体目标包括&#xff1a; &#xff08;1&#xff09;掌握网络拓扑的…

记一次:使用C#创建一个串口工具

前言&#xff1a;公司的上位机打不开串口&#xff0c;发送的时候设备总是关机&#xff0c;因为和这个同事关系比较好&#xff0c;编写这款软件是用C#编写的&#xff0c;于是乎帮着解决了一下&#xff08;是真解决了&#xff09;&#xff0c;然后整理了一下自己的笔记 一、开发…

SQL 获取今天的当月开始结束范围:

使用 GETDATE() 结合 DATEADD() 和 DATEDIFF() 函数来获取当前月的开始和结束时间范围。以下是实现当前月时间范围查询的 SQL&#xff1a; FDATE > DATEADD(MONTH, DATEDIFF(MONTH, 0, GETDATE()), 0) FDATE < DATEADD(MONTH, DATEDIFF(MONTH, 0, GETDATE()) 1, 0) …

利用Java爬虫MinC根据ID获取商品详情的完整指南

在当今数字化时代&#xff0c;获取商品详情数据对于市场分析、价格监控和竞争对手分析至关重要。Java作为一种强大且广泛使用的编程语言&#xff0c;非常适合开发复杂的爬虫系统。本文将详细介绍如何利用Java编写爬虫程序来根据商品ID获取商品详情&#xff0c;并提供完整的代码…

RabbitMQ-入门

RabbitMQ-入门 基本说明 AMQP协议&#xff08;高消息队列协议&#xff1a;Advanced Message Queuing Protocol&#xff09;&#xff1a;https://www.rabbitmq.com/tutorials/amqp-concepts.html 基础组件术语 生产者&#xff08;Publisher&#xff09;&#xff1a;发消息到某…

【JAVA项目】基于ssm的【美食推荐管理系统】

【JAVA项目】基于ssm的【美食推荐管理系统】 技术简介&#xff1a;采用JSP技术、B/S架构、SSM框架、MySQL技术等实现。 系统简介&#xff1a;美食推荐管理系统&#xff0c;在系统首页可以查看首页、热门美食、美食教程、美食店铺、美食社区、美食资讯、我的、跳转到后台等内容。…

go-zero(十二)消息队列

go zero 消息队列 在微服务架构中&#xff0c;消息队列主要通过异步通信实现服务间的解耦&#xff0c;使得各个服务可以独立发展和扩展。 go-zero中使用的队列组件go-queue&#xff0c;是gozero官方实现的基于Kafka和Beanstalkd 的消息队列框架,我们使用kafka作为演示。 一、…

day30-IO-阶段综合案例(带权重的随机每日一记)(笔记完全来源于黑马程序员)

目录 0 目录一、听黑马阿玮的视频记录的笔记1. 制造假数据1.1 如何制造假数据1.2 练习1-生成方式1&#xff1a;爬取姓氏、男生名字、女生名字1.3 练习2-生成方式1&#xff1a;在练习1的基础上&#xff0c;将数据写入本地文件1.4 练习3-生成方式2&#xff1a;利用糊涂包生成假数…

Excel的文件导入遇到大文件时

Excel的文件导入向导如何把已导入数据排除 入起始行&#xff0c;选择从哪一行开始导入。 比如&#xff0c;前两行已经导入了&#xff0c;第二次导入的时候排除前两行&#xff0c;从第三行开始&#xff0c;就将导入起始行设置为3即可&#xff0c;且不勾选含标题行。 但遇到大文…

qemu安装arm64架构银河麒麟

qemu虚拟化软件&#xff0c;可以在一个平台上模拟另一个硬件平台&#xff0c;可以支持多种处理器架构。 一、安装 安装教程&#xff1a;https://blog.csdn.net/qq_36035382/article/details/125308044 下载链接&#xff1a;https://qemu.weilnetz.de/w64/2024/ 我下载的是 …

Linux-ubuntu环境配置

一&#xff0c;安装VWware&#xff0c;里面导入镜像文件 这些都是文件夹里面有的&#xff0c;然后对着正点原子视频安装就行&#xff0c;虚拟机的破解码&#xff0c;去百度搜一个能用就行&#xff0c;中间遇见俩问题。①乌班图里面不能上网&#xff0c;②插入U盘后&#xff0c;…

如何在 Ubuntu 22.04 上安装和使用 Apache Kafka

简介 Apache Kafka是一个高性能、低延迟的分布式流处理平台&#xff0c;广泛用于构建实时数据管道和流式应用。本文将指导你如何在Ubuntu 22.04系统上快速部署Apache Kafka&#xff0c;让你体验到Kafka在处理大规模实时数据流方面的强大能力。通过本教程&#xff0c;你将学会如…

TCP/IP杂记

TCP三次握手、四次挥手 从应用角度&#xff0c;不用多考虑为什么有三次&#xff0c;遵循标准即可。 ubuntu 下 wireshark安装&#xff1a; sudo add-apt-repository universe sudo apt install wireshark 三次握手实证&#xff1a; 第一次握手的情况如下&#xff1a;&#…

【2024最新Java面试宝典】—— SpringBoot面试题(44道含答案)

1. 什么是 Spring Boot&#xff1f; Spring Boot 是 Spring 开源组织下的子项目&#xff0c;是 Spring 组件一站式解决方案&#xff0c;主要是简化了使用 Spring 的难度&#xff0c;简省了繁重的配置&#xff0c;提供了各种启动器&#xff0c;使开发者能快速上手。 2. 为什么…

MTK Android12 开机向导

文章目录 需求-场景参考资料&#xff1a;博客资料官网参考资料&#xff1a;参考资料注意点 附件资料文件说明&#xff1a;推荐工具&#xff1a;配置定制的 声明叠加层 APK需求实现替换字符、删减开机向导界面、添加开机向导界面删除部分界面需求&#xff0c;官网说明如下更新部…

思科模拟器路由器的基本配置

一、实验目的 了解路由器的作用掌握路由器的基本配置方法 3、掌握路由器模块的使用和互连方式 二、实验环境 2811路由器一台&#xff0c;计算机两台&#xff0c;Console配置线一根&#xff0c;网线若干&#xff1b;本实验拓扑图如图8-1所示&#xff1b;计算机IP地址规划如表8-…