Table Topic 和腾讯云 Iceberg 集成

​

前置条件

• 版本要求： AutoMQ 实例版本为 1.4.1 或更高版本。
• Catalog 要求： 准备可访问的 Hive Metastore 服务。你可以自行部署 Hive Metastore，也可以使用腾讯云 EMR 提供的 HMS。
• 存储要求： 提前创建 COS Bucket，并确认 AutoMQ 实例所在地域与 Bucket 地域一致。记下包含 APPID 后缀的完整 Bucket 标识、地域和 endpoint，例如 examplebucket-1250000000、ap-shanghai 和 https://cos.ap-shanghai.myqcloud.com。
• 权限要求： 为 AutoMQ 实例使用的 CAM Role 授予访问 COS Bucket 的权限。创建 Hive Catalog 集成时，控制台会生成所需的 Policy。

​

操作步骤

​

步骤 1：准备 Hive Metastore 和 COS

​

配置 EMR HMS

1. 前往腾讯云 EMR 产品，创建或选择一个包含 Hive Metastore 的 EMR 集群。如果创建新集群，请选择 Hive 组件；后续查询还需要 Spark 组件。根据你的安全要求决定是否开启 Kerberos。

1. 在 EMR 控制台进入集群服务 > HIVE > 配置管理，选择 hive-env.sh，点击编辑配置，添加访问对象存储所需的 Jar。以下配置适用于本文示例环境；如果 EMR 版本不同，请根据实际 Hadoop 版本调整 Jar 路径和版本。

# 添加 Jar 依赖，让 HMS 支持访问对象存储
# hive-env.sh
HIVE_AUX_JARS_PATH=/opt/apps/HADOOP-COMMON/hadoop-current/share/hadoop/tools/lib/aws-java-sdk-bundle-1.11.375.jar:/opt/apps/HADOOP-COMMON/hadoop-current/share/hadoop/tools/lib/hadoop-aws-3.2.1.jar

1. 在 HMS 使用的 hive-site.xml 中配置 COS 的 S3 兼容访问参数。以下示例使用静态凭证访问 COS，请将 YOUR_TENCENT_SECRET_ID 和 YOUR_TENCENT_SECRET_KEY 替换为具备目标 Bucket 访问权限的实际凭证。

# hive-site.xml
fs.s3a.endpoint=https://cos.ap-shanghai.myqcloud.com
fs.s3a.access.key=YOUR_TENCENT_SECRET_ID
fs.s3a.secret.key=YOUR_TENCENT_SECRET_KEY
fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem

1. 保存配置并重启 Hive Metastore，使配置生效。

​

准备 Kerberos 文件（可选）

kadmin.local
addprinc -pw YOUR_PASSWORD automq-user@YOUR_REALM
ktadd -k /root/automq-user.keytab automq-user@YOUR_REALM
quit

cp /etc/krb5.conf /root/krb5.conf

• hive.metastore.uris：Hive Metastore 接入点，格式为 thrift://<host>:<port>。它填入 AutoMQ 的 Hive Metastore Endpoint，不是 User Principal。
• hive.metastore.kerberos.principal：Hive Metastore 服务端 principal，填入 AutoMQ 的 Kerberos Principal。
• automq-user@YOUR_REALM：上面创建的客户端 principal，填入 AutoMQ 的 Kerberos User Principal。

​

步骤 2：创建 Hive Catalog 集成

1. 登录 AutoMQ 控制台，进入集成页面。

1. 选择创建 Hive Catalog 集成，填写以下信息：
   • 名称： 填写便于识别的集成名称。
   • 部署配置： 选择 AutoMQ 实例所属的部署配置。该配置必须与后续创建的实例一致。
   • Hive Metastore Endpoint： 填写步骤 1 中获取的 hive.metastore.uris，例如 thrift://hms.example.com:7004。
   • 鉴权模式： 根据 HMS 配置选择无认证或 Kerberos。
   • Keytab 文件： 使用 Kerberos 时，上传客户端 keytab 文件。
   • krb5.conf 文件： 使用 Kerberos 时，上传 krb5.conf 文件。
   • Kerberos User Principal： 使用 Kerberos 时，填写客户端 principal，例如 automq-user@YOUR_REALM。
   • Kerberos Principal： 使用 Kerberos 时，填写 hive.metastore.kerberos.principal 的值，即 HMS 服务端 principal。
   • Warehouse： 填写包含 APPID 后缀的完整 COS Bucket 标识，不要填写 s3a:// 前缀。AutoMQ 会使用 <bucket>/iceberg 作为 Iceberg Warehouse 路径。

1. 复制控制台根据 Warehouse 生成的 COS Policy，在腾讯云 CAM 控制台创建自定义策略，并将策略附加到 AutoMQ 实例使用的 CAM Role。至少确认该 Role 对目标 Bucket 及其对象拥有 Table Topic 所需的读、写、列举和删除权限。
2. 完成授权后，创建 Hive Catalog 集成。

​

步骤 3：创建实例并开启 Table Topic

1. 开启 Table Topic。
2. 选择 Hive Catalog。
3. 选择步骤 2 创建的 Hive Catalog 集成。

​

步骤 4：创建 Topic 并配置流转表

1. 进入实例的 Topics 页面，点击创建 Topic。
2. 配置 Topic 的基础参数：
   • Topic 名称： 填写 Topic 名称，该名称也会作为 Iceberg 表名。
3. 开启 Table Topic 转换，并配置以下参数：
   • 命名空间： Iceberg Catalog 中的 Database 名称，用于隔离不同的 Iceberg 表。
   • Schema 约束类型：
     • Schema： 消息必须遵守已注册的 Schema。需要配置AutoMQ 内置的 Schema Registry并注册消息 Schema；Table Topic 使用 Schema 的字段创建和填充 Iceberg 表。
     • Schemaless： 消息没有明确的 Schema，Table Topic 将消息 Key 和 Value 作为整体字段写入 Iceberg 表。
4. 点击确定创建 Topic。

​

步骤 5：生产消息并使用 Spark SQL 查询

1. 打开 Topic 详情中的生产消息页签，输入测试消息的 Key 和 Value，然后发送消息。
2. 确认 EMR 集群包含 Spark 组件；如果创建集群时未选择 Spark，请先添加 Spark 组件。登录 EMR Master 节点并切换到 hadoop 用户。
3. 本示例使用截图中的 Spark 3.2.2、Scala 2.12 和 Iceberg 0.13.1。若 EMR 使用其他版本，请同步调整以下变量，确保 iceberg-spark-runtime 的 Spark/Scala 后缀与实际环境匹配。

export HIVE_URI="thrift://YOUR_HMS_HOST:YOUR_HMS_PORT"
export COS_BUCKET="YOUR_BUCKET_APPID"
export COS_REGION="ap-shanghai"
export COS_ACCESS_KEY="YOUR_TENCENT_SECRET_ID"
export COS_SECRET_KEY="YOUR_TENCENT_SECRET_KEY"
export COS_ENDPOINT="https://cos.${COS_REGION}.myqcloud.com"
export ICEBERG_SPARK_RUNTIME="3.2_2.12"
export ICEBERG_VERSION="0.13.1"
export ICEBERG_AWS="org.apache.iceberg:iceberg-aws:${ICEBERG_VERSION}"
export ICEBERG_RUNTIME="org.apache.iceberg:iceberg-spark-runtime-${ICEBERG_SPARK_RUNTIME}:${ICEBERG_VERSION}"

spark-sql --master local[*] \
  --packages "${ICEBERG_RUNTIME},${ICEBERG_AWS}" \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.defaultCatalog=hive \
  --conf spark.sql.catalog.hive=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.hive.type=hive \
  --conf spark.sql.catalog.hive.uri="${HIVE_URI}" \
  --conf spark.sql.catalog.hive.warehouse="s3a://${COS_BUCKET}/iceberg" \
  --conf spark.sql.catalog.hive.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
  --conf spark.sql.catalog.hive.s3.endpoint="${COS_ENDPOINT}" \
  --conf spark.sql.catalog.hive.client.region="${COS_REGION}" \
  --conf spark.sql.catalog.hive.s3.path-style-access=false \
  --conf spark.sql.catalog.hive.s3.access-key-id="${COS_ACCESS_KEY}" \
  --conf spark.sql.catalog.hive.s3.secret-access-key="${COS_SECRET_KEY}" \
  --conf spark.hadoop.fs.s3a.endpoint="${COS_ENDPOINT}" \
  --conf spark.hadoop.fs.s3a.region="${COS_REGION}" \
  --conf spark.hadoop.fs.s3a.path.style.access=false \
  --conf spark.hadoop.fs.s3a.access.key="${COS_ACCESS_KEY}" \
  --conf spark.hadoop.fs.s3a.secret.key="${COS_SECRET_KEY}" \
  --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem

4. 在 spark-sql 中查询表。将 <namespace> 和 <topic> 替换为创建 Topic 时使用的命名空间和 Topic 名称：

SHOW NAMESPACES IN hive;
SHOW TABLES IN hive.<namespace>;
SELECT * FROM hive.<namespace>.<topic> LIMIT 20;