lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2045] Support Read Hoodie As DataSource Table For Flink And DeltaStreamer

Browse Source
This commit is contained in:
pengzhiwei
2021-06-21 14:13:25 +08:00
parent 5804ad8e32
commit ffa934182a
14 changed files with 470 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java
View File

@@ -110,6 +110,12 @@ public class HiveSyncConfig implements Serializable {
@Parameter(names = {"--batch-sync-num"}, description = "The number of partitions one batch when synchronous partitions to hive")
public Integer batchSyncNum = 1000;
@Parameter(names = {"--spark-datasource"}, description = "Whether sync this table as spark data source table.")
public Boolean syncAsSparkDataSourceTable = true;
@Parameter(names = {"--spark-schema-length-threshold"}, description = "The maximum length allowed in a single cell when storing additional schema information in Hive's metastore.")
public int sparkSchemaLengthThreshold = 4000;
// enhance the similar function in child class
public static HiveSyncConfig copy(HiveSyncConfig cfg) {
HiveSyncConfig newConfig = new HiveSyncConfig();
@@ -131,6 +137,8 @@ public class HiveSyncConfig implements Serializable {
newConfig.serdeProperties = cfg.serdeProperties;
newConfig.createManagedTable = cfg.createManagedTable;
newConfig.batchSyncNum = cfg.batchSyncNum;
newConfig.syncAsSparkDataSourceTable = cfg.syncAsSparkDataSourceTable;
newConfig.sparkSchemaLengthThreshold = cfg.sparkSchemaLengthThreshold;
return newConfig;
}
@@ -160,6 +168,8 @@ public class HiveSyncConfig implements Serializable {
+ ", supportTimestamp=" + supportTimestamp
+ ", decodePartition=" + decodePartition
+ ", createManagedTable=" + createManagedTable
+ ", syncAsSparkDataSourceTable=" + syncAsSparkDataSourceTable
+ ", sparkSchemaLengthThreshold=" + sparkSchemaLengthThreshold
+ '}';
}
}

113
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
View File

@@ -20,11 +20,13 @@ package org.apache.hudi.hive;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.InvalidTableException;
import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
import org.apache.hudi.hive.util.ConfigUtils;
import org.apache.hudi.hive.util.Parquet2SparkSchemaUtils;
import org.apache.hudi.sync.common.AbstractSyncHoodieClient.PartitionEvent;
import org.apache.hudi.sync.common.AbstractSyncHoodieClient.PartitionEvent.PartitionEventType;
import org.apache.hudi.hive.util.HiveSchemaUtil;
@@ -37,13 +39,20 @@ import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hudi.sync.common.AbstractSyncTool;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.parquet.schema.OriginalType.UTF8;
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
/**
* Tool to sync a hoodie HDFS table with a hive metastore table. Either use it as a api
* HiveSyncTool.syncHoodieTable(HiveSyncConfig) or as a command line java -cp hoodie-hive-sync.jar HiveSyncTool [args]
@@ -152,6 +161,16 @@ public class HiveSyncTool extends AbstractSyncTool {
// Get the parquet schema for this table looking at the latest commit
MessageType schema = hoodieHiveClient.getDataSchema();
// Currently HoodieBootstrapRelation does support reading bootstrap MOR rt table,
// so we disable the syncAsSparkDataSourceTable here to avoid read such kind table
// by the data source way (which will use the HoodieBootstrapRelation).
// TODO after we support bootstrap MOR rt table in HoodieBootstrapRelation[HUDI-2071], we can remove this logical.
if (hoodieHiveClient.isBootstrap()
&& hoodieHiveClient.getTableType() == HoodieTableType.MERGE_ON_READ
&& !readAsOptimized) {
cfg.syncAsSparkDataSourceTable = false;
}
// Sync schema if needed
syncSchema(tableName, tableExists, useRealtimeInputFormat, readAsOptimized, schema);
@@ -180,6 +199,15 @@ public class HiveSyncTool extends AbstractSyncTool {
*/
private void syncSchema(String tableName, boolean tableExists, boolean useRealTimeInputFormat,
boolean readAsOptimized, MessageType schema) {
// Append spark table properties & serde properties
Map<String, String> tableProperties = ConfigUtils.toMap(cfg.tableProperties);
Map<String, String> serdeProperties = ConfigUtils.toMap(cfg.serdeProperties);
if (cfg.syncAsSparkDataSourceTable) {
Map<String, String> sparkTableProperties = getSparkTableProperties(cfg.sparkSchemaLengthThreshold, schema);
Map<String, String> sparkSerdeProperties = getSparkSerdeProperties(readAsOptimized);
tableProperties.putAll(sparkTableProperties);
serdeProperties.putAll(sparkSerdeProperties);
}
// Check and sync schema
if (!tableExists) {
LOG.info("Hive table " + tableName + " is not found. Creating it");
@@ -196,27 +224,11 @@ public class HiveSyncTool extends AbstractSyncTool {
String outputFormatClassName = HoodieInputFormatUtils.getOutputFormatClassName(baseFileFormat);
String serDeFormatClassName = HoodieInputFormatUtils.getSerDeClassName(baseFileFormat);
Map<String, String> serdeProperties = ConfigUtils.toMap(cfg.serdeProperties);
// The serdeProperties is non-empty only for spark sync meta data currently.
if (!serdeProperties.isEmpty()) {
String queryTypeKey = serdeProperties.remove(ConfigUtils.SPARK_QUERY_TYPE_KEY);
String queryAsROKey = serdeProperties.remove(ConfigUtils.SPARK_QUERY_AS_RO_KEY);
String queryAsRTKey = serdeProperties.remove(ConfigUtils.SPARK_QUERY_AS_RT_KEY);
if (queryTypeKey != null && queryAsROKey != null && queryAsRTKey != null) {
if (readAsOptimized) { // read optimized
serdeProperties.put(queryTypeKey, queryAsROKey);
} else { // read snapshot
serdeProperties.put(queryTypeKey, queryAsRTKey);
}
}
}
// Custom serde will not work with ALTER TABLE REPLACE COLUMNS
// https://github.com/apache/hive/blob/release-1.1.0/ql/src/java/org/apache/hadoop/hive
// /ql/exec/DDLTask.java#L3488
hoodieHiveClient.createTable(tableName, schema, inputFormatClassName,
outputFormatClassName, serDeFormatClassName, serdeProperties, ConfigUtils.toMap(cfg.tableProperties));
outputFormatClassName, serDeFormatClassName, serdeProperties, tableProperties);
} else {
// Check if the table schema has evolved
Map<String, String> tableSchema = hoodieHiveClient.getTableSchema(tableName);
@@ -226,7 +238,6 @@ public class HiveSyncTool extends AbstractSyncTool {
hoodieHiveClient.updateTableDefinition(tableName, schema);
// Sync the table properties if the schema has changed
if (cfg.tableProperties != null) {
Map<String, String> tableProperties = ConfigUtils.toMap(cfg.tableProperties);
hoodieHiveClient.updateTableProperties(tableName, tableProperties);
LOG.info("Sync table properties for " + tableName + ", table properties is: " + cfg.tableProperties);
}
@@ -236,6 +247,72 @@ public class HiveSyncTool extends AbstractSyncTool {
}
}
/**
* Get Spark Sql related table properties. This is used for spark datasource table.
* @param schema The schema to write to the table.
* @return A new parameters added the spark's table properties.
*/
private Map<String, String> getSparkTableProperties(int schemaLengthThreshold, MessageType schema) {
// Convert the schema and partition info used by spark sql to hive table properties.
// The following code refers to the spark code in
// https://github.com/apache/spark/blob/master/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
GroupType originGroupType = schema.asGroupType();
List<String> partitionNames = cfg.partitionFields;
List<Type> partitionCols = new ArrayList<>();
List<Type> dataCols = new ArrayList<>();
Map<String, Type> column2Field = new HashMap<>();
for (Type field : originGroupType.getFields()) {
column2Field.put(field.getName(), field);
}
// Get partition columns and data columns.
for (String partitionName : partitionNames) {
// Default the unknown partition fields to be String.
// Keep the same logical with HiveSchemaUtil#getPartitionKeyType.
partitionCols.add(column2Field.getOrDefault(partitionName,
new PrimitiveType(Type.Repetition.REQUIRED, BINARY, partitionName, UTF8)));
}
for (Type field : originGroupType.getFields()) {
if (!partitionNames.contains(field.getName())) {
dataCols.add(field);
}
}
List<Type> reOrderedFields = new ArrayList<>();
reOrderedFields.addAll(dataCols);
reOrderedFields.addAll(partitionCols);
GroupType reOrderedType = new GroupType(originGroupType.getRepetition(), originGroupType.getName(), reOrderedFields);
Map<String, String> sparkProperties = new HashMap<>();
sparkProperties.put("spark.sql.sources.provider", "hudi");
// Split the schema string to multi-parts according the schemaLengthThreshold size.
String schemaString = Parquet2SparkSchemaUtils.convertToSparkSchemaJson(reOrderedType);
int numSchemaPart = (schemaString.length() + schemaLengthThreshold - 1) / schemaLengthThreshold;
sparkProperties.put("spark.sql.sources.schema.numParts", String.valueOf(numSchemaPart));
// Add each part of schema string to sparkProperties
for (int i = 0; i < numSchemaPart; i++) {
int start = i * schemaLengthThreshold;
int end = Math.min(start + schemaLengthThreshold, schemaString.length());
sparkProperties.put("spark.sql.sources.schema.part." + i, schemaString.substring(start, end));
}
// Add partition columns
if (!partitionNames.isEmpty()) {
sparkProperties.put("spark.sql.sources.schema.numPartCols", String.valueOf(partitionNames.size()));
for (int i = 0; i < partitionNames.size(); i++) {
sparkProperties.put("spark.sql.sources.schema.partCol." + i, partitionNames.get(i));
}
}
return sparkProperties;
}
private Map<String, String> getSparkSerdeProperties(boolean readAsOptimized) {
Map<String, String> sparkSerdeProperties = new HashMap<>();
sparkSerdeProperties.put("path", cfg.basePath);
sparkSerdeProperties.put(ConfigUtils.IS_QUERY_AS_RO_TABLE, String.valueOf(readAsOptimized));
return sparkSerdeProperties;
}
/**
* Syncs the list of storage parititions passed in (checks if the partition is in hive, if not adds it or if the
* partition path does not match, it updates the partition path).

11
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ConfigUtils.java
View File

@@ -23,12 +23,11 @@ import java.util.Map;
import org.apache.hudi.common.util.StringUtils;
public class ConfigUtils {
public static final String SPARK_QUERY_TYPE_KEY = "spark.query.type.key";
public static final String SPARK_QUERY_AS_RO_KEY = "spark.query.as.ro.key";
public static final String SPARK_QUERY_AS_RT_KEY = "spark.query.as.rt.key";
/**
* Config stored in hive serde properties to tell query engine (spark/flink) to
* read the table as a read-optimized table when this config is true.
*/
public static final String IS_QUERY_AS_RO_TABLE = "hoodie.query.as.ro.table";
/**
* Convert the key-value config to a map.The format of the config

171
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/Parquet2SparkSchemaUtils.java Normal file
View File

@@ -0,0 +1,171 @@
/*
* 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.hive.util;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.OriginalType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
/**
* Convert the parquet schema to spark schema' json string.
* This code is refer to org.apache.spark.sql.execution.datasources.parquet.ParquetToSparkSchemaConverter
* in spark project.
*/
public class Parquet2SparkSchemaUtils {
public static String convertToSparkSchemaJson(GroupType parquetSchema) {
String fieldsJsonString = parquetSchema.getFields().stream().map(field -> {
switch (field.getRepetition()) {
case OPTIONAL:
return "{\"name\":\"" + field.getName() + "\",\"type\":" + convertFieldType(field)
+ ",\"nullable\":true,\"metadata\":{}}";
case REQUIRED:
return "{\"name\":\"" + field.getName() + "\",\"type\":" + convertFieldType(field)
+ ",\"nullable\":false,\"metadata\":{}}";
case REPEATED:
String arrayType = arrayType(field, false);
return "{\"name\":\"" + field.getName() + "\",\"type\":" + arrayType
+ ",\"nullable\":false,\"metadata\":{}}";
default:
throw new UnsupportedOperationException("Unsupport convert " + field + " to spark sql type");
}
}).reduce((a, b) -> a + "," + b).orElse("");
return "{\"type\":\"struct\",\"fields\":[" + fieldsJsonString + "]}";
}
private static String convertFieldType(Type field) {
if (field instanceof PrimitiveType) {
return "\"" + convertPrimitiveType((PrimitiveType) field) + "\"";
} else {
assert field instanceof GroupType;
return convertGroupField((GroupType) field);
}
}
private static String convertPrimitiveType(PrimitiveType field) {
PrimitiveType.PrimitiveTypeName typeName = field.getPrimitiveTypeName();
OriginalType originalType = field.getOriginalType();
switch (typeName) {
case BOOLEAN: return "boolean";
case FLOAT: return "float";
case DOUBLE: return "double";
case INT32:
if (originalType == null) {
return "integer";
}
switch (originalType) {
case INT_8: return "byte";
case INT_16: return "short";
case INT_32: return "integer";
case DATE: return "date";
case DECIMAL:
return "decimal(" + field.getDecimalMetadata().getPrecision() + ","
+ field.getDecimalMetadata().getScale() + ")";
default: throw new UnsupportedOperationException("Unsupport convert " + typeName + " to spark sql type");
}
case INT64:
if (originalType == null) {
return "long";
}
switch (originalType) {
case INT_64: return "long";
case DECIMAL:
return "decimal(" + field.getDecimalMetadata().getPrecision() + ","
+ field.getDecimalMetadata().getScale() + ")";
case TIMESTAMP_MICROS:
case TIMESTAMP_MILLIS:
return "timestamp";
default:
throw new UnsupportedOperationException("Unsupport convert " + typeName + " to spark sql type");
}
case INT96: return "timestamp";
case BINARY:
if (originalType == null) {
return "binary";
}
switch (originalType) {
case UTF8:
case ENUM:
case JSON:
return "string";
case BSON: return "binary";
case DECIMAL:
return "decimal(" + field.getDecimalMetadata().getPrecision() + ","
+ field.getDecimalMetadata().getScale() + ")";
default:
throw new UnsupportedOperationException("Unsupport convert " + typeName + " to spark sql type");
}
case FIXED_LEN_BYTE_ARRAY:
switch (originalType) {
case DECIMAL:
return "decimal(" + field.getDecimalMetadata().getPrecision() + ","
+ field.getDecimalMetadata().getScale() + ")";
default:
throw new UnsupportedOperationException("Unsupport convert " + typeName + " to spark sql type");
}
default:
throw new UnsupportedOperationException("Unsupport convert " + typeName + " to spark sql type");
}
}
private static String convertGroupField(GroupType field) {
if (field.getOriginalType() == null) {
return convertToSparkSchemaJson(field);
}
switch (field.getOriginalType()) {
case LIST:
ValidationUtils.checkArgument(field.getFieldCount() == 1, "Illegal List type: " + field);
Type repeatedType = field.getType(0);
if (isElementType(repeatedType, field.getName())) {
return arrayType(repeatedType, false);
} else {
Type elementType = repeatedType.asGroupType().getType(0);
boolean optional = elementType.isRepetition(OPTIONAL);
return arrayType(elementType, optional);
}
case MAP:
case MAP_KEY_VALUE:
GroupType keyValueType = field.getType(0).asGroupType();
Type keyType = keyValueType.getType(0);
Type valueType = keyValueType.getType(1);
boolean valueOptional = valueType.isRepetition(OPTIONAL);
return "{\"type\":\"map\", \"keyType\":" + convertFieldType(keyType)
+ ",\"valueType\":" + convertFieldType(valueType)
+ ",\"valueContainsNull\":" + valueOptional + "}";
default:
throw new UnsupportedOperationException("Unsupport convert " + field + " to spark sql type");
}
}
private static String arrayType(Type elementType, boolean containsNull) {
return "{\"type\":\"array\", \"elementType\":" + convertFieldType(elementType) + ",\"containsNull\":" + containsNull + "}";
}
private static boolean isElementType(Type repeatedType, String parentName) {
return repeatedType.isPrimitive() || repeatedType.asGroupType().getFieldCount() > 1
|| repeatedType.getName().equals("array") || repeatedType.getName().equals(parentName + "_tuple");
}
}

74
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
View File

@@ -42,7 +42,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
@@ -70,6 +69,11 @@ public class TestHiveSyncTool {
return Arrays.asList(new Object[][] {{true, true, true}, {true, false, false}, {false, true, true}, {false, false, false}});
}
// (useJdbc, useSchemaFromCommitMetadata, syncAsDataSource)
private static Iterable<Object[]> syncDataSourceTableParams() {
return Arrays.asList(new Object[][] {{true, true, true}, {true, false, false}, {false, true, true}, {false, false, false}});
}
@BeforeEach
public void setUp() throws Exception {
HiveTestUtil.setUp();
@@ -157,17 +161,15 @@ public class TestHiveSyncTool {
}
@ParameterizedTest
@MethodSource({"useJdbcAndSchemaFromCommitMetadata"})
@MethodSource({"syncDataSourceTableParams"})
public void testSyncCOWTableWithProperties(boolean useJdbc,
boolean useSchemaFromCommitMetadata) throws Exception {
boolean useSchemaFromCommitMetadata,
boolean syncAsDataSourceTable) throws Exception {
HiveSyncConfig hiveSyncConfig = HiveTestUtil.hiveSyncConfig;
HiveTestUtil.hiveSyncConfig.batchSyncNum = 3;
Map<String, String> serdeProperties = new HashMap<String, String>() {
{
put("path", hiveSyncConfig.basePath);
put(ConfigUtils.SPARK_QUERY_TYPE_KEY, "hoodie.datasource.query.type");
put(ConfigUtils.SPARK_QUERY_AS_RO_KEY, "read_optimized");
put(ConfigUtils.SPARK_QUERY_AS_RT_KEY, "snapshot");
}
};
@@ -177,6 +179,7 @@ public class TestHiveSyncTool {
put("tp_1", "p1");
}
};
hiveSyncConfig.syncAsSparkDataSourceTable = syncAsDataSourceTable;
hiveSyncConfig.useJdbc = useJdbc;
hiveSyncConfig.serdeProperties = ConfigUtils.configToString(serdeProperties);
hiveSyncConfig.tableProperties = ConfigUtils.configToString(tableProperties);
@@ -195,9 +198,12 @@ public class TestHiveSyncTool {
String tblPropertiesWithoutDdlTime = String.join("\n",
results.subList(0, results.size() - 1));
String sparkTableProperties = getSparkTableProperties(syncAsDataSourceTable, useSchemaFromCommitMetadata);
assertEquals(
"EXTERNAL\tTRUE\n"
+ "last_commit_time_sync\t100\n"
+ sparkTableProperties
+ "tp_0\tp0\n"
+ "tp_1\tp1", tblPropertiesWithoutDdlTime);
assertTrue(results.get(results.size() - 1).startsWith("transient_lastDdlTime"));
@@ -208,21 +214,54 @@ public class TestHiveSyncTool {
hiveDriver.getResults(results);
String ddl = String.join("\n", results);
assertTrue(ddl.contains("'path'='" + hiveSyncConfig.basePath + "'"));
assertTrue(ddl.contains("'hoodie.datasource.query.type'='snapshot'"));
if (syncAsDataSourceTable) {
assertTrue(ddl.contains("'" + ConfigUtils.IS_QUERY_AS_RO_TABLE + "'='false'"));
}
}
private String getSparkTableProperties(boolean syncAsDataSourceTable, boolean useSchemaFromCommitMetadata) {
if (syncAsDataSourceTable) {
if (useSchemaFromCommitMetadata) {
return "spark.sql.sources.provider\thudi\n"
+ "spark.sql.sources.schema.numPartCols\t1\n"
+ "spark.sql.sources.schema.numParts\t1\n"
+ "spark.sql.sources.schema.part.0\t{\"type\":\"struct\",\"fields\":"
+ "[{\"name\":\"_hoodie_commit_time\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},"
+ "{\"name\":\"_hoodie_commit_seqno\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},"
+ "{\"name\":\"_hoodie_record_key\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},"
+ "{\"name\":\"_hoodie_partition_path\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},"
+ "{\"name\":\"_hoodie_file_name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},"
+ "{\"name\":\"name\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}},"
+ "{\"name\":\"favorite_number\",\"type\":\"integer\",\"nullable\":false,\"metadata\":{}},"
+ "{\"name\":\"favorite_color\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}},"
+ "{\"name\":\"datestr\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}}]}\n"
+ "spark.sql.sources.schema.partCol.0\tdatestr\n";
} else {
return "spark.sql.sources.provider\thudi\n"
+ "spark.sql.sources.schema.numPartCols\t1\n"
+ "spark.sql.sources.schema.numParts\t1\n"
+ "spark.sql.sources.schema.part.0\t{\"type\":\"struct\",\"fields\":[{\"name\":\"name\",\"type\":"
+ "\"string\",\"nullable\":false,\"metadata\":{}},{\"name\":\"favorite_number\",\"type\":\"integer\","
+ "\"nullable\":false,\"metadata\":{}},{\"name\":\"favorite_color\",\"type\":\"string\",\"nullable\":false,"
+ "\"metadata\":{}}]}\n"
+ "{\"name\":\"datestr\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}}]}\n"
+ "spark.sql.sources.schema.partCol.0\tdatestr\n";
}
} else {
return "";
}
}
@ParameterizedTest
@MethodSource({"useJdbcAndSchemaFromCommitMetadata"})
@MethodSource({"syncDataSourceTableParams"})
public void testSyncMORTableWithProperties(boolean useJdbc,
boolean useSchemaFromCommitMetadata) throws Exception {
boolean useSchemaFromCommitMetadata,
boolean syncAsDataSourceTable) throws Exception {
HiveSyncConfig hiveSyncConfig = HiveTestUtil.hiveSyncConfig;
HiveTestUtil.hiveSyncConfig.batchSyncNum = 3;
Map<String, String> serdeProperties = new HashMap<String, String>() {
{
put("path", hiveSyncConfig.basePath);
put(ConfigUtils.SPARK_QUERY_TYPE_KEY, "hoodie.datasource.query.type");
put(ConfigUtils.SPARK_QUERY_AS_RO_KEY, "read_optimized");
put(ConfigUtils.SPARK_QUERY_AS_RT_KEY, "snapshot");
}
};
@@ -232,6 +271,7 @@ public class TestHiveSyncTool {
put("tp_1", "p1");
}
};
hiveSyncConfig.syncAsSparkDataSourceTable = syncAsDataSourceTable;
hiveSyncConfig.useJdbc = useJdbc;
hiveSyncConfig.serdeProperties = ConfigUtils.configToString(serdeProperties);
hiveSyncConfig.tableProperties = ConfigUtils.configToString(tableProperties);
@@ -247,14 +287,15 @@ public class TestHiveSyncTool {
String rtTableName = hiveSyncConfig.tableName + HiveSyncTool.SUFFIX_SNAPSHOT_TABLE;
String[] tableNames = new String[] {roTableName, rtTableName};
String[] expectQueryTypes = new String[] {"read_optimized", "snapshot"};
String[] readAsOptimizedResults = new String[] {"true", "false"};
SessionState.start(HiveTestUtil.getHiveConf());
Driver hiveDriver = new org.apache.hadoop.hive.ql.Driver(HiveTestUtil.getHiveConf());
String sparkTableProperties = getSparkTableProperties(syncAsDataSourceTable, useSchemaFromCommitMetadata);
for (int i = 0;i < 2; i++) {
String dbTableName = hiveSyncConfig.databaseName + "." + tableNames[i];
String expectQueryType = expectQueryTypes[i];
String readAsOptimized = readAsOptimizedResults[i];
hiveDriver.run("SHOW TBLPROPERTIES " + dbTableName);
List<String> results = new ArrayList<>();
@@ -265,6 +306,7 @@ public class TestHiveSyncTool {
assertEquals(
"EXTERNAL\tTRUE\n"
+ "last_commit_time_sync\t101\n"
+ sparkTableProperties
+ "tp_0\tp0\n"
+ "tp_1\tp1", tblPropertiesWithoutDdlTime);
assertTrue(results.get(results.size() - 1).startsWith("transient_lastDdlTime"));
@@ -275,8 +317,10 @@ public class TestHiveSyncTool {
hiveDriver.getResults(results);
String ddl = String.join("\n", results);
assertTrue(ddl.contains("'path'='" + hiveSyncConfig.basePath + "'"));
assertTrue(ddl.contains("'hoodie.datasource.query.type'='" + expectQueryType + "'"));
assertTrue(ddl.toLowerCase().contains("create external table"));
if (syncAsDataSourceTable) {
assertTrue(ddl.contains("'" + ConfigUtils.IS_QUERY_AS_RO_TABLE + "'='" + readAsOptimized + "'"));
}
}
}

84
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestParquet2SparkSchemaUtils.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* 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.hive;
import org.apache.hudi.hive.util.Parquet2SparkSchemaUtils;
import org.apache.spark.sql.execution.SparkSqlParser;
import org.apache.spark.sql.execution.datasources.parquet.SparkToParquetSchemaConverter;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.types.ArrayType;
import org.apache.spark.sql.types.MapType;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.IntegerType$;
import org.apache.spark.sql.types.StringType$;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class TestParquet2SparkSchemaUtils {
private final SparkToParquetSchemaConverter spark2ParquetConverter =
new SparkToParquetSchemaConverter(
(Boolean) SQLConf.PARQUET_WRITE_LEGACY_FORMAT().defaultValue().get(),
SQLConf.ParquetOutputTimestampType$.MODULE$.INT96());
private final SparkSqlParser parser = new SparkSqlParser(new SQLConf());
@Test
public void testConvertPrimitiveType() {
StructType sparkSchema = parser.parseTableSchema(
"f0 int, f1 string, f3 bigint,"
+ " f4 decimal(5,2), f5 timestamp, f6 date,"
+ " f7 short, f8 float, f9 double, f10 byte,"
+ " f11 tinyint, f12 smallint, f13 binary, f14 boolean");
String sparkSchemaJson = Parquet2SparkSchemaUtils.convertToSparkSchemaJson(
spark2ParquetConverter.convert(sparkSchema).asGroupType());
StructType convertedSparkSchema = (StructType) StructType.fromJson(sparkSchemaJson);
assertEquals(sparkSchema.json(), convertedSparkSchema.json());
// Test type with nullable
StructField field0 = new StructField("f0", StringType$.MODULE$, false, Metadata.empty());
StructField field1 = new StructField("f1", StringType$.MODULE$, true, Metadata.empty());
StructType sparkSchemaWithNullable = new StructType(new StructField[]{field0, field1});
String sparkSchemaWithNullableJson = Parquet2SparkSchemaUtils.convertToSparkSchemaJson(
spark2ParquetConverter.convert(sparkSchemaWithNullable).asGroupType());
StructType convertedSparkSchemaWithNullable = (StructType) StructType.fromJson(sparkSchemaWithNullableJson);
assertEquals(sparkSchemaWithNullable.json(), convertedSparkSchemaWithNullable.json());
}
@Test
public void testConvertComplexType() {
StructType sparkSchema = parser.parseTableSchema(
"f0 int, f1 map<string, int>, f2 array<decimal(10,2)>"
+ ",f3 map<array<date>, bigint>, f4 array<array<double>>"
+ ",f5 struct<id:int, name:string>");
String sparkSchemaJson = Parquet2SparkSchemaUtils.convertToSparkSchemaJson(
spark2ParquetConverter.convert(sparkSchema).asGroupType());
StructType convertedSparkSchema = (StructType) StructType.fromJson(sparkSchemaJson);
assertEquals(sparkSchema.json(), convertedSparkSchema.json());
// Test complex type with nullable
StructField field0 = new StructField("f0", new ArrayType(StringType$.MODULE$, true), false, Metadata.empty());
StructField field1 = new StructField("f1", new MapType(StringType$.MODULE$, IntegerType$.MODULE$, true), false, Metadata.empty());
StructType sparkSchemaWithNullable = new StructType(new StructField[]{field0, field1});
String sparkSchemaWithNullableJson = Parquet2SparkSchemaUtils.convertToSparkSchemaJson(
spark2ParquetConverter.convert(sparkSchemaWithNullable).asGroupType());
StructType convertedSparkSchemaWithNullable = (StructType) StructType.fromJson(sparkSchemaWithNullableJson);
assertEquals(sparkSchemaWithNullable.json(), convertedSparkSchemaWithNullable.json());
}
}