Skip to content

[HUDI-3978] Fix use of partition path field as hive partition field in flink #5434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yihua merged 3 commits into from
Apr 30, 2022
Merged

[HUDI-3978] Fix use of partition path field as hive partition field in flink #5434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9908f6b
fix partition path fields as hive sync partition fields error
onlywangyh Apr 28, 2022
9e20555
Update TestHiveSyncContext.java
onlywangyh Apr 29, 2022
8f9e73d
Merge branch 'apache:master' into HUDI-3962
onlywangyh Apr 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions ...flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.hudi.sink.utils;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.hudi.aws.sync.AwsGlueCatalogSyncTool;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.configuration.FlinkOptions;
Expand Down Expand Up @@ -70,7 +71,8 @@ public static HiveSyncContext create(Configuration conf) {
return new HiveSyncContext(syncConfig, hiveConf, fs);
}

private static HiveSyncConfig buildSyncConfig(Configuration conf) {
@VisibleForTesting
public static HiveSyncConfig buildSyncConfig(Configuration conf) {
HiveSyncConfig hiveSyncConfig = new HiveSyncConfig();
hiveSyncConfig.basePath = conf.getString(FlinkOptions.PATH);
hiveSyncConfig.baseFileFormat = conf.getString(FlinkOptions.HIVE_SYNC_FILE_FORMAT);
Expand All @@ -83,7 +85,7 @@ private static HiveSyncConfig buildSyncConfig(Configuration conf) {
hiveSyncConfig.tableProperties = conf.getString(FlinkOptions.HIVE_SYNC_TABLE_PROPERTIES);
hiveSyncConfig.serdeProperties = conf.getString(FlinkOptions.HIVE_SYNC_TABLE_SERDE_PROPERTIES);
hiveSyncConfig.jdbcUrl = conf.getString(FlinkOptions.HIVE_SYNC_JDBC_URL);
hiveSyncConfig.partitionFields = Arrays.asList(FilePathUtils.extractPartitionKeys(conf));
hiveSyncConfig.partitionFields = Arrays.asList(FilePathUtils.extractHivePartitionFields(conf));
hiveSyncConfig.partitionValueExtractorClass = conf.getString(FlinkOptions.HIVE_SYNC_PARTITION_EXTRACTOR_CLASS_NAME);
hiveSyncConfig.useJdbc = conf.getBoolean(FlinkOptions.HIVE_SYNC_USE_JDBC);
hiveSyncConfig.useFileListingFromMetadata = conf.getBoolean(FlinkOptions.METADATA_ENABLED);
Expand Down
13 changes: 13 additions & 0 deletions ...flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FilePathUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,4 +425,17 @@ public static String[] extractPartitionKeys(org.apache.flink.configuration.Confi
}
return conf.getString(FlinkOptions.PARTITION_PATH_FIELD).split(",");
}

/**
* Extracts the hive sync partition fields with given configuration.
*
* @param conf The flink configuration
* @return array of the hive partition fields
*/
public static String[] extractHivePartitionFields(org.apache.flink.configuration.Configuration conf) {
if (FlinkOptions.isDefaultValueDefined(conf, FlinkOptions.HIVE_SYNC_PARTITION_FIELDS)) {
return extractPartitionKeys(conf);
}
return conf.getString(FlinkOptions.HIVE_SYNC_PARTITION_FIELDS).split(",");
}
Copy link
Contributor

@danny0405 danny0405 Apr 26, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In HiveSyncContext, we already assign the partition fields to to hive sync partition fields, see:

hudi/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java

Line 86 in 762623a

hiveSyncConfig.partitionFields = Arrays.asList(FilePathUtils.extractPartitionKeys(conf));

so what's the problem here ? BTW, this is a common util which you should not modify directly.

Copy link
Contributor Author

@onlywangyh onlywangyh Apr 26, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In HiveSyncContext this PARTITION_PATH_FIELD assign to hive sync partition fields .I think these two params PARTITION_PATH_FIELD and HIVE_SYNC_PARTITION_FIELDS
have different meanings in hudi.
PARTITION_PATH_FIELD is for hudi KeyGenerator to get a partitionPath
HIVE_SYNC_PARTITION_FIELDS is use for hive to set a partition field.

This function extractPartitionKeys should get the hive partition fields key rather than a hudi partition path field. Sometimes confuse the values ​ will cause some errors

In this case we use TimestampBasedAvroKeyGenerator and set hudi partition path field is same as hive partition fields . There will be some promblems, see:
PARTITION_PATH_FIELD=datetime HIVE_SYNC_PARTITION_FIELDS=datetime

In hudi: we will get the 1596074902000L value and converted to a string hudi partition path like 2020-07-30.
In hive: We will get the table like :

 CREATE EXTERNAL TABLE `testTable`(
   `_hoodie_commit_time` string COMMENT '',       
   `_hoodie_commit_seqno` string COMMENT '',          
   `_hoodie_record_key` string COMMENT '',
   `_hoodie_partition_path` string COMMENT '',
   `_hoodie_file_name` string COMMENT '',               
   `id` int COMMENT '',
   `datetime` bigint COMMENT ''        
   )
 PARTITIONED BY (`datetime` string COMMENT '')...

This partition value datetime=2020-07-30 also will be add to hive. We can't get the datetime value from this hive table, and the table partition is also broken. This datetime value is conflicting

When we set PARTITION PATH_FIELD value is different with HIVE_SYNC PARTITION FIELDS value like this.
PARTITION_PATH_FIELD=datetime HIVE_SYNC_PARTITION_FIELDS=inc_day
We can get this table like ：

CREATE EXTERNAL TABLE `testTable`(
   `_hoodie_commit_time` string COMMENT '',       
   `_hoodie_commit_seqno` string COMMENT '',          
   `_hoodie_record_key` string COMMENT '',
   `_hoodie_partition_path` string COMMENT '',
   `_hoodie_file_name` string COMMENT '',               
   `id` int COMMENT '',
   `datetime` bigint COMMENT ''        
   )
 PARTITIONED BY (`inc_day` string COMMENT '')...

In this time we can normal get the datetime value, and the inc_day as a partition field is also work.

Copy link
Contributor

@danny0405 danny0405 Apr 27, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got you, then in HiveSyncContext, you can take the option HIVE_SYNC_PARTITION_FIELDS instead, and write a new tool method(you should not modify the existing method directly), if user does not configure the HIVE_SYNC_PARTITION_FIELDS, fallback to hudi table's partition path (the current code does so).

Copy link
Contributor

@danny0405 danny0405 Apr 27, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You also need to write a test case i guess. A unit test for HiveSyncContext is okey i think.

}
62 changes: 62 additions & 0 deletions ...k-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestHiveSyncContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.hudi.sink.utils;

import org.apache.flink.configuration.Configuration;

import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.hive.HiveSyncConfig;

import org.junit.jupiter.api.Test;

import java.lang.reflect.Method;

import static org.junit.jupiter.api.Assertions.assertTrue;

/**
* Test cases for {@link HiveSyncContext}.
*/
public class TestHiveSyncContext {
/**
* Test that the file ids generated by the task can finally shuffled to itself.
*/
@Test
public void testBuildSyncConfig() throws Exception {
Configuration configuration1 = new Configuration();
Configuration configuration2 = new Configuration();
String hiveSyncPartitionField = "hiveSyncPartitionField";
String partitionPathField = "partitionPathField";

configuration1.setString(FlinkOptions.HIVE_SYNC_PARTITION_FIELDS, hiveSyncPartitionField);
configuration1.setString(FlinkOptions.PARTITION_PATH_FIELD, partitionPathField);

configuration2.setString(FlinkOptions.PARTITION_PATH_FIELD, partitionPathField);

Class<?> threadClazz = Class.forName("org.apache.hudi.sink.utils.HiveSyncContext");
Method buildSyncConfigMethod = threadClazz.getDeclaredMethod("buildSyncConfig", Configuration.class);
buildSyncConfigMethod.setAccessible(true);

HiveSyncConfig hiveSyncConfig1 = HiveSyncContext.buildSyncConfig(configuration1);
HiveSyncConfig hiveSyncConfig2 = HiveSyncContext.buildSyncConfig(configuration2);

assertTrue(hiveSyncConfig1.partitionFields.get(0).equals(hiveSyncPartitionField));
assertTrue(hiveSyncConfig2.partitionFields.get(0).equals(partitionPathField));

}
}