[SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
**_Tips before filing an issue_**
- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?
Yes
- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.
- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.
**Describe the problem you faced**
I have a Flink job (1.16) writing a Hudi COW table with async clustering. The way I am writing the job is:
* Using the Table API to create the table
* Subsequently doing an insert into the table
```scala
// In Flink Scala
val hudiTable = "my_table_name"
val hudiTableDescriptor = buildHudiTableDescriptor()
logger.info(s"Creating hudi table {} with descriptor: {}", hudiTable, hudiTableDescriptor.toString())
bsTableEnv.createTable(hudiTable, hudiTableDescriptor)
bsTableEnv.fromDataStream(d).select(getOrderedColumnExpr(): _*).insertInto(AppConfig.output.hudiSink.targetTable).execute()
```
The table is registered as an external table in Hive, but Flink is not attached to any Hive catalogue. The table is registered in SparkSQL using:
```
spark.sql("create table xxx.yyy using hudi location 's3a://location_flink_is_writing_to' ").show()
```
Right now, I want to try schema evolution and to add a column to the table. I am creating a savepoint, adding the column into the Flink Hudi DDL, and recreating the table, and subsequently inserting with the new column.
On the Spark side, I have tried:
1. Dropping the external table and recreating it after resuming my flink job from the savepoint with new columns (it still doesn't register the new column, probably because hoodie.properties is not being updated by Flink)
2. Stop my job, trying to `ALTER TABLE ADD COLUMN ` from spark following https://hudi.apache.org/docs/schema_evolution/, i.e.:
```
spark.sql("alter table xxx.yyy add columns (new_column ARRAY)").show(100, False)
```
and resuming my job from the savepoint but with the new columns above.
Option (1) doesn't end up registering a new column.
Option (2) the column is added in the hive metastore but all values are still null, even new incoming rows.
**To Reproduce**
Steps to reproduce the behavior:
1. Create a Flink job writing data to s3, by creating a hudi table with the table API and then doing an insert
2. Register the table in Hive as an external table
4. Restart the Flink job from a savepoint, adding in the new column into the DDL to run the create again and insert
5. Try to add the column by either recreating the external table or by doing an `alter table add column` in Spark SQL
6. Query table using Spark SQL using the hive table
7. All nulls returned for new column
```
spark.sql("""
SELECT request_id, new_column
from xxx.yyy
where concat(local_year,local_month, local_day) = '20230721'
and new_column is not null
limit 100
""").show(100, False)
+----------+-------------------+
|request_id|new_column |
+----------+-------------------+
+----------+-------------------+
```
**Expected behavior**
Expected to be able to add a column successfully through either method and for column to be populated with data and queryable.
**Environment Description**
* Hudi version : 0.13
* Spark version : 3.3
* Hive version : 2
* Hadoop version:
* Storage (HDFS/S3/GCS..) : S3
* Running on Docker? (yes/no) : Yes, on Kubernetes
**Additional context**
spark configuration
```
conf = pyspark.SparkConf()\
.set("spark.executor.instances", number_of_executors)\
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")\
.set("spark.sql.catalogImplementation", "hive")\
.set("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")\
.set("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")\
.set("spark.jars.packages", "org.apache.hudi:hudi-spark3.3-bundle_2.12:0.13.1")\
.set("spark.hadoop.fs.auth.grabfs.detailed.error.msg", "true")\
.set("spark.hadoop.fs.s3a.multiobjectdelete.enable", "false")
```
flink hudi table configuration:
```
WITH (
'hoodie.clustering.plan.strategy.daybased.lookback.partitions' = '2',
'compaction.schedule.enabled' = 'true',
'clustering.plan.strategy.sort.columns' = 'cola,colb,colc,cold',
'hoodie.datasource.write.reconcile.schema' = 'true',
'hoodie.write.lock.provider' = 'org.apache.hudi.client.transaction.lock.InProcessLockProvider',
'hoodie.schema.on.read.enable' = 'true',
'clustering.plan.strategy.small.file.limit' = '100',
'hoodie.parquet.max.file.size' = '26214400',
'path' = 's3a://location_flink_is_writing_to',
'hoodie.datasource.write.hive_style_partitioning' = 'true',
'hoodie.cleaner.hours.retained' = '1',
'write.merge.max_memory' = '512',
'hoodie.parquet.small.file.limit' = '13107200',
'hoodie.cleaner.policy' = 'KEEP_LATEST_BY_HOURS',
'write.operation' = 'insert',
'write.insert.cluster' = 'false',
'table.table' = 'COPY_ON_WRITE',
'precombine.field' = 'event_datetime_local',
'write.task.max.size' = '1024',
'hoodie.clustering.async.max.commits' = '1',
'hoodie.write.concurrency.mode' = 'optimistic_concurrency_control',
'clustering.tasks' = '30',
'connector' = 'hudi',
'clustering.async.enabled' = 'true',
'hoodie.cleaner.policy.failed.writes' = 'LAZY',
'clustering.plan.partition.filter.mode' = 'RECENT_DAYS',
'hoodie.copyonwrite.record.size.estimate' = '104857600',
'clustering.schedule.enabled' = 'true',
'write.sort.memory' = '512'
)
```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with the Flink Table API createTable and insert flow described in the issue, then compare it with the Spark SQL external-table and ALTER TABLE paths. Reproduce the savepoint restart and query steps against the S3 table; done means the added column is registered and populated for new rows rather than returning only nulls.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, docker, kubernetes, python, scala
- Domain
- cloud, data-engineering, databases, stream-processing
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 20/100