29、Spark写数据到Hudi时，同步hive表的一些坑

1.hudi的同步hive表没有comment

原以为hudi同步的hive表是根据数据写入的dataframe的schema创建的。就和spark write hive时类似，查察源码后发现不是。
1.1 hudi同步hive的模式

HMS , JDBC , HIVESQL。我这儿常用的是HMS和JDBC

各个同步模式对应的实行器：

1.2 schema天生

我们可以看到schema天生的代码块。先从提交的commit中获取元数据信息，没有的话则从数据文件中获取schema。两种方式获取到的schema都是没有comment信息的。
org.apache.hudi.common.table.TableSchemaResolver#getTableParquetSchema

1.   /**
2.    * Gets the schema for a hoodie table. Depending on the type of table, read from any file written in the latest
3.    * commit. We will assume that the schema has not changed within a single atomic write.
4.    *
5.    * @return Parquet schema for this table
6.    * @throws Exception
7.    */
8.   private MessageType getTableParquetSchemaFromDataFile() throws Exception {
9.     HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
11.     try {
12.       switch (metaClient.getTableType()) {
13.         case COPY_ON_WRITE:
14.           // If this is COW, get the last commit and read the schema from a file written in the
15.           // last commit
16.           HoodieInstant lastCommit =
17.               activeTimeline.getCommitsTimeline().filterCompletedInstants().lastInstant().orElseThrow(() -> new InvalidTableException(metaClient.getBasePath()));
18.           HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
19.               .fromBytes(activeTimeline.getInstantDetails(lastCommit).get(), HoodieCommitMetadata.class);
20.           String filePath = commitMetadata.getFileIdAndFullPaths(metaClient.getBasePath()).values().stream().findAny()
21.               .orElseThrow(() -> new IllegalArgumentException("Could not find any data file written for commit "
22.                   + lastCommit + ", could not get schema for table " + metaClient.getBasePath() + ", Metadata :"
23.                   + commitMetadata));
24.           return readSchemaFromBaseFile(new Path(filePath));
25.         case MERGE_ON_READ:
26.           // If this is MOR, depending on whether the latest commit is a delta commit or
27.           // compaction commit
28.           // Get a datafile written and get the schema from that file
29.           Option<HoodieInstant> lastCompactionCommit =
30.               metaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().lastInstant();
31.           LOG.info("Found the last compaction commit as " + lastCompactionCommit);
33.           Option<HoodieInstant> lastDeltaCommit;
34.           if (lastCompactionCommit.isPresent()) {
35.             lastDeltaCommit = metaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants()
36.                 .findInstantsAfter(lastCompactionCommit.get().getTimestamp(), Integer.MAX_VALUE).lastInstant();
37.           } else {
38.             lastDeltaCommit =
39.                 metaClient.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().lastInstant();
40.           }
41.           LOG.info("Found the last delta commit " + lastDeltaCommit);
43.           if (lastDeltaCommit.isPresent()) {
44.             HoodieInstant lastDeltaInstant = lastDeltaCommit.get();
45.             // read from the log file wrote
46.             commitMetadata = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(lastDeltaInstant).get(),
47.                 HoodieCommitMetadata.class);
48.             Pair<String, HoodieFileFormat> filePathWithFormat =
49.                 commitMetadata.getFileIdAndFullPaths(metaClient.getBasePath()).values().stream()
50.                     .filter(s -> s.contains(HoodieLogFile.DELTA_EXTENSION)).findAny()
51.                     .map(f -> Pair.of(f, HoodieFileFormat.HOODIE_LOG)).orElseGet(() -> {
52.                       // No Log files in Delta-Commit. Check if there are any parquet files
53.                       return commitMetadata.getFileIdAndFullPaths(metaClient.getBasePath()).values().stream()
54.                           .filter(s -> s.contains((metaClient.getTableConfig().getBaseFileFormat().getFileExtension())))
55.                           .findAny().map(f -> Pair.of(f, HoodieFileFormat.PARQUET)).orElseThrow(() ->
56.                               new IllegalArgumentException("Could not find any data file written for commit "
57.                               + lastDeltaInstant + ", could not get schema for table " + metaClient.getBasePath()
58.                               + ", CommitMetadata :" + commitMetadata));
59.                     });
60.             switch (filePathWithFormat.getRight()) {
61.               case HOODIE_LOG:
62.                 return readSchemaFromLogFile(lastCompactionCommit, new Path(filePathWithFormat.getLeft()));
63.               case PARQUET:
64.                 return readSchemaFromBaseFile(new Path(filePathWithFormat.getLeft()));
65.               default:
66.                 throw new IllegalArgumentException("Unknown file format :" + filePathWithFormat.getRight()
67.                     + " for file " + filePathWithFormat.getLeft());
68.             }
69.           } else {
70.             return readSchemaFromLastCompaction(lastCompactionCommit);
71.           }
72.         default:
73.           LOG.error("Unknown table type " + metaClient.getTableType());
74.           throw new InvalidTableException(metaClient.getBasePath());
75.       }
76.     } catch (IOException e) {
77.       throw new HoodieException("Failed to read data schema", e);
78.     }
79.   }

复制代码

1.3建表DDL

获取到schema后，我们再看建表行为。
org.apache.hudi.hive.ddl.DDLExecutor#createTable 定义了这个接口建表方法。有两个实现类，一个是
org.apache.hudi.hive.ddl.HMSDDLExecutor。另一个是 org.apache.hudi.hive.ddl.QueryBasedDDLExecutor

首先，看org.apache.hudi.hive.ddl.HMSDDLExecutor#createTable方法：
ddl操作中利用的字段信息在HiveSchemaUtil.convertMapSchemaToHiveFieldSchema天生，可以直接在这个方法里看到字段的comment信息是直接写死为空字符串的。

再看，org.apache.hudi.hive.ddl.QueryBasedDDLExecutor#createTable方法。
方法里是通过HiveSchemaUtil.generateCreateDDL方法直接天生的ddl建表语句的。这个方法里generateSchemaString方法来天生字段信息的。在这个方法里，也是没有涉及comment信息的。

1.4结论

同步hive表是在 数据写入hudi目次后，根据目次里的schema来创建的hive表，所以创建的hive表没有带着dataframe的comment信息。需要手动实行修改字段comment。
2.追加comment

2.1.利用spark.sql的方式修改comment

用spark.sql(）的方式实行 修改comment的sql语句，会调用hudi里的AlterHoodieTableChangeColumnCommand类。这个里面会比较schema，革新sparksession里的catalog信息，会让使命hang住。（为什么hang住没去排查）大概操作就是写一个利用新的schema的空数据集到hudi来实现schema更新。
org.apache.spark.sql.hudi.command.AlterHoodieTableChangeColumnCommand。

2.2利用hive-sql的方式修改comment

用hive-jdbc的方式实行修改sql语句。这个方式不会更新hive表里的 TBLPROPERTIES 的 'spark.sql.sources.schema.part.0’信息。
利用dataframe的schame.tojson ，去修改 ‘spark.sql.sources.schema.part.0’ 信息

1.   /**
2.    * 将 dataframe 中的comment加到 hudi的hive表中
3.    *
4.    * @param df      dataframe
5.    * @param dbTable hive表
6.    * @param spark   spark session
7.    */
8.   def addCommentForSyncHive(df: DataFrame, dbTable: String, spark: SparkSession, writeOptions: mutable.Map[String, String]): Unit = {
9.     val comment: Map[String, String] = df.schema.map(sf => (sf.name, sf.getComment().getOrElse(""))).toMap
10.     info(s"数据集的字段名->备注为：\n${comment.mkString("\n")}")
13.     val jdbcUrlOption = writeOptions.get(DataSourceWriteOptions.HIVE_URL.key())
14.     val jdbcUserOption = writeOptions.get(DataSourceWriteOptions.HIVE_USER.key())
15.     val jdbcPassOption = writeOptions.get(DataSourceWriteOptions.HIVE_PASS.key())
16.     assert(jdbcUrlOption.isDefined, s"${DataSourceWriteOptions.HIVE_URL.key()} 必须被指定")
18.     val connection = DbUtil.createHiveConnection(
19.       jdbcUrlOption.get, jdbcUserOption.getOrElse(""), jdbcPassOption.getOrElse("")
20.     )
21.     val stmt = connection.createStatement()
22.     //需要手动更新hive表中的spark.sql.sources.schema.part.0信息
23.     stmt.execute(s"ALTER TABLE $dbTable SET TBLPROPERTIES ('spark.sql.sources.schema.part.0' = '${df.schema.json}')")
25.     // 获取表字段和类型
26.     val tableSchema = spark.sql(s"DESCRIBE $dbTable")
27.       .select("col_name", "data_type")
28.       .collect()
29.       .map(row => (row.getString(0), row.getString(1)))
31.     tableSchema.foreach { case (column, dataType) =>
32.       if (comment.contains(column) && !Seq("ym", "ymd").contains(column)) {
33.         val newComment = comment.getOrElse(column, "")
34.         val sql = s"""ALTER TABLE $dbTable CHANGE COLUMN $column $column $dataType COMMENT '$newComment'"""
35.         info(s"添加备注执行sql：$sql")
36.         try {
37.           stmt.execute(sql)
38.         } catch {
39.           case e:Throwable =>
40.             warn("添加备注sql执行失败")
41.         }
42.       }
43.     }
44.     stmt.close()
45.     connection.close()
46.   }

复制代码

修改’spark.sql.sources.schema.part.0’时，因为schema带有备注，会很长，导致超过hive表元数据mysql表字段的长度限制。去mysql中修改这个长度限制（table_params表PARAM_VALUE字段）。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。