You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "jlloh (via GitHub)" <gi...@apache.org> on 2023/07/21 10:13:14 UTC
[GitHub] [hudi] jlloh opened a new issue, #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
jlloh opened a new issue, #9256:
URL: https://github.com/apache/hudi/issues/9256
**_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 compaction. 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<STRING>)").show(100, False)
```
and resuming my job from the savepoint but with the new columns above.
(1) doesn't end up registering a new column.
(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' = 'country_code,card_bin,ip_address,passenger_id,request_id,booking_code,rule_id,version_id',
'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'
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] voonhous commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1654921822
TBH, that's a pretty tricky question to answer, not sure how to answer it as i feel it's pretty dependent on a few factors. Source + Hudi-on-Flink job type.
Maybe @danny0405 can chime in on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] voonhous commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1646777044
@jlloh Oh, i meant the latest snapshot on the current master branch. Alright, will check in on this tomorrow again!
LMK again. Thanks for responding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] jlloh commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "jlloh (via GitHub)" <gi...@apache.org>.
jlloh commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1651495991
> P.S. Can you please help to check if the configs below contain the new column spec definition that was added via Spark?
```
getOrderedColumnExpr()
AppConfig.output.hudiSink.targetTable
```
`AppConfig.output.hudiSink.targetTable` is a string of the temp table that the Flink append only job writes to. Using your example, it would be `wt_test_hudi`.
`getOrderedColumnExpr()` generates a select expression in flink to ensure the order of the columns inserted into the hudi table in Flink. I.e. `select col_1, col_2, col_2`, so it should definitely contain the new column as well.
One thing I noticed differs in my configuration vs yours is that my Flink table is not connected to the Hive metastore:
```
'hive_sync.enable'='true',
'hive_sync.metastore.uris' = 'thrift://hive.metastore.io:9083',
'hive_sync.use_jdbc' = 'false',
'hive_sync.mode' = 'hms',
'hive_sync.db' = 'dev_hudi',
'hive_sync.table' = 'hfse_flink_writes'
```
All these would be false.
When I did my alter table add partition, I did not set the session property, but instead I did:
```
alter = """
alter table my_hive_schema.table SET TBLPROPERTIES ('hoodie.metadata.enable' = 'false', 'hoodie.schema.on.read.enable' = 'true')
"""
spark.sql(alter).show()
```
Before I did a alter table add partition. Not sure if this is equivalent.
Also, one thing I noticed is that `hoodie.properties` does not change to reflect the presence of my new columns. Let me know if you still want me to test a primitive, non-nested column.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] jlloh commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "jlloh (via GitHub)" <gi...@apache.org>.
jlloh commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1653434166
Update, @voonhous, turns out it was indeed an issue with my Flink setup. It was setup in Kubernetes in HA mode, and I was creating a savepoint, deleting the pods, and resuming by it by passing it the argument `-s`.
But it seems that in HA mode, somehow if the configmap in Kubernetes exists, it will ignore the savepoint and load from the checkpoint instead, and it was not picking up the new columns I added to my job, so the new columns were not being inserted.
It worked as expected with this sequence:
1. create savepoint
2. delete flink deployment, clean up HA configmap
3. `alter table add column` in spark
4. redeploy new flink job with new column, and make sure that the new jobgraph for flink is updated to insert the new column
5. query new column from spark
Thanks again for the quick support @voonhous, apologies that it was a misconfiguration on my side.
p.s. My minor question still stands above on starting a flink job from scratch and writing to the same hudi table, if you know the answer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] jlloh commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "jlloh (via GitHub)" <gi...@apache.org>.
jlloh commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1652989720
Ok so I tried loading in the specific parquet files written in partitions whereby the new column should exist, and realised that the parquet files don't contain the new column. Seems it might be an issue specific to my Flink job, when I reload from the savepoint it's not picking up the new column or modifying the jobgraph. Will dig into this further.
One follow up question slightly off topic, what's the "safe" way to restart a Flink job from scratch and get it to write to the same Hudi table? I.e. let's say it's a problem with updating the jobgraph from the savepoint, and I want to stop my Flink job with some data loss, and continue writing to the hudi table after redeploying. I could not find much documentation on how I can let my Flink job just resume writing to an existing Hudi table and ignore the previous in-flight commits that did not finish.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] voonhous commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1646559079
Question, is this issue only applicable to new `ARRAY<XXX>` types?
Can you adding a column with a primitive type first and see if the same issue persist.
I would also suggest trying to see if this issue can be replicated with the latest version with option(2).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] jlloh commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "jlloh (via GitHub)" <gi...@apache.org>.
jlloh commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1646699216
Hey @voonhous I am using 0.13.1 actually, is that what you mean by latest version?
Sure let me try with a primitive type sometime next week but I suspect when I recreate the table flink might not update the table properties or data since I do see a log along the lines of `skipping update of table properties because table exists ` or something like that.
I have existing columns of complex types and those work fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] voonhous commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1651552403
> Also, one thing I noticed is that hoodie.properties does not change to reflect the presence of my new columns. Let me know if you still want me to test a primitive, non-nested column.
This is expected, hoodie.properties will not change. The metadata containing the schema is situated in:
```
hdfs://my/path/to/hfse_flink_writes/.hoodie/.schema
```
> One thing I noticed differs in my configuration vs yours is that my Flink table is not connected to the Hive metastore:
I am not using hive-catalogue, the hive-sync configs are only used for sync-ing after every writes.
I did the test again without hive-catalogue + hive sync, same results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] voonhous commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1647280925
@jlloh I'm able to write and newly added column with such an operational logic:
# Create a Hudi-on-Flink append only job
```sql
CREATE TABLE input_table (
`val` STRING
,`event_time` TIMESTAMP(3)
,`partition` BIGINT
,`offset` BIGINT
,`map_col` MAP<STRING, INT>
) WITH (
'connector' = 'datagen',
'fields.val.length' = '99',
'rows-per-second' = '50'
);
CREATE TABLE wt_test_hudi
(
`val` STRING
,`event_time` TIMESTAMP(3)
,`partition` BIGINT
,`offset` BIGINT
,`map_col` MAP<STRING, INT>
,`dt` STRING
) PARTITIONED BY (dt)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://my/path/to/hfse_flink_writes',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'insert',
'hoodie.parquet.small.file.limit' = '104857600',
'hoodie.parquet.max.file.size' = '268435456',
'hoodie.datasource.write.recordkey.field' = 'val',
'hoodie.datasource.write.hive_style_partitioning' = 'true',
'hoodie.datasource.write.partitionpath.field' = 'dt',
'hoodie.datasource.write.keygenerator.class' = 'org.apache.hudi.keygen.ComplexAvroKeyGenerator',
'write.tasks' = '2',
'clustering.schedule.enabled' = 'true',
'clustering.async.enabled' = 'true',
'clustering.plan.strategy.small.file.limit' = '1800', -- 1800MB
'clustering.plan.strategy.target.file.max.bytes' = '2147483648', -- 2GB
'clustering.tasks' = '5',
'clustering.delta_commits' = '5',
'hoodie.parquet.compression.ratio' = '0.1',
'hive_sync.enable'='true',
'hive_sync.metastore.uris' = 'thrift://hive.metastore.io:9083',
'hive_sync.use_jdbc' = 'false',
'hive_sync.mode' = 'hms',
'hive_sync.db' = 'dev_hudi',
'hive_sync.table' = 'hfse_flink_writes'
);
insert into wt_test_hudi
select `val`
,`event_time`
,`partition`
,`offset`
, `map_col`
, `array_col`
,DATE_FORMAT(event_time, 'yyyy-MM-dd')
from input_table;
```
# Stop the flink job (with or without savepoint) + alter table on spark
On a spark session, update the table schema
```sql
ALTER TABLE dev_hudi.hfse_flink_writes ADD COLUMN (`array_col` ARRAY<INT> AFTER `map_col`);
```
This will add a the column `array_col` after `map_col`.
# Update Hudi-on-Flink append-only job with the new schema
Update the Flink-SQL and start the job again.
```sql
CREATE TABLE input_table (
`val` STRING
,`event_time` TIMESTAMP(3)
,`partition` BIGINT
,`offset` BIGINT
,`map_col` MAP<STRING, INT>
,`array_col` ARRAY< INT>
) WITH (
'connector' = 'datagen',
'fields.val.length' = '99',
'rows-per-second' = '50'
);
CREATE TABLE wt_test_hudi
(
`val` STRING
,`event_time` TIMESTAMP(3)
,`partition` BIGINT
,`offset` BIGINT
,`map_col` MAP<STRING, INT>
,`array_col` ARRAY< INT>
,`dt` STRING
) PARTITIONED BY (dt)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://my/path/to/hfse_flink_writes',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'insert',
'hoodie.parquet.small.file.limit' = '104857600',
'hoodie.parquet.max.file.size' = '268435456',
'hoodie.datasource.write.recordkey.field' = 'val',
'hoodie.datasource.write.hive_style_partitioning' = 'true',
'hoodie.datasource.write.partitionpath.field' = 'dt',
'hoodie.datasource.write.keygenerator.class' = 'org.apache.hudi.keygen.ComplexAvroKeyGenerator',
'write.tasks' = '2',
'clustering.schedule.enabled' = 'true',
'clustering.async.enabled' = 'true',
'clustering.plan.strategy.small.file.limit' = '1800', -- 1800MB
'clustering.plan.strategy.target.file.max.bytes' = '2147483648', -- 2GB
'clustering.tasks' = '5',
'clustering.delta_commits' = '5',
'hoodie.parquet.compression.ratio' = '0.1',
'hive_sync.enable'='true',
'hive_sync.metastore.uris' = 'thrift://hive.metastore.io:9083',
'hive_sync.use_jdbc' = 'false',
'hive_sync.mode' = 'hms',
'hive_sync.db' = 'dev_hudi',
'hive_sync.table' = 'hfse_flink_writes'
);
insert into wt_test_hudi
select `val`
,`event_time`
,`partition`
,`offset`
, `map_col`
, `array_col`
,DATE_FORMAT(event_time, 'yyyy-MM-dd')
from input_table;
```
# Query the table results on Spark
```sql
REFRESH TABLE dev_hudi.hfse_flink_writes;
SELECT * FROM dev_hudi.hfse_flink_writes WHERE `array_col` is not null;
```
# PS
Can you please help to check if the configs below contain the new column spec definition that was added via Spark?
1. getOrderedColumnExpr()
2. AppConfig.output.hudiSink.targetTable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] jlloh commented on issue #9256: [SUPPORT] How to do schema evolution for Flink Hudi table registered on Hive
Posted by "jlloh (via GitHub)" <gi...@apache.org>.
jlloh commented on issue #9256:
URL: https://github.com/apache/hudi/issues/9256#issuecomment-1648874648
Sure let me get back to you sometime later this week when I get some time to test your suggestions. Thanks for the quick support.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org