JingsongLi commented on a change in pull request #9864: [FLINK-14254][table]
Introduce FileSystemOutputFormat for batch
URL: https://github.com/apache/flink/pull/9864#discussion_r349986703
##########
File path:
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
##########
@@ -86,46 +83,36 @@ public HiveTableSink(JobConf jobConf, ObjectPath
tablePath, CatalogTable table)
@Override
public OutputFormat<Row> getOutputFormat() {
- List<String> partitionColumns = getPartitionFieldNames();
- boolean isPartitioned = partitionColumns != null &&
!partitionColumns.isEmpty();
- boolean isDynamicPartition = isPartitioned &&
partitionColumns.size() > staticPartitionSpec.size();
+ String[] partitionColumns =
getPartitionFieldNames().toArray(new String[0]);
String dbName = tablePath.getDatabaseName();
String tableName = tablePath.getObjectName();
try (HiveMetastoreClientWrapper client =
HiveMetastoreClientFactory.create(new HiveConf(jobConf, HiveConf.class),
hiveVersion)) {
Table table = client.getTable(dbName, tableName);
StorageDescriptor sd = table.getSd();
- // here we use the sdLocation to store the output path
of the job, which is always a staging dir
- String sdLocation = sd.getLocation();
- HiveTablePartition hiveTablePartition;
- if (isPartitioned) {
- validatePartitionSpec();
- if (isDynamicPartition) {
- List<String> path = new ArrayList<>(2);
- path.add(sd.getLocation());
- if (!staticPartitionSpec.isEmpty()) {
-
path.add(Warehouse.makePartName(staticPartitionSpec, false));
- }
- sdLocation =
String.join(Path.SEPARATOR, path);
- } else {
- List<Partition> partitions =
client.listPartitions(dbName, tableName,
- new
ArrayList<>(staticPartitionSpec.values()), (short) 1);
- sdLocation = !partitions.isEmpty() ?
partitions.get(0).getSd().getLocation() :
- sd.getLocation() +
Path.SEPARATOR + Warehouse.makePartName(staticPartitionSpec, true);
- }
-
- sd.setLocation(toStagingDir(sdLocation,
jobConf));
- hiveTablePartition = new HiveTablePartition(sd,
new LinkedHashMap<>(staticPartitionSpec));
- } else {
- sd.setLocation(toStagingDir(sdLocation,
jobConf));
- hiveTablePartition = new HiveTablePartition(sd,
null);
- }
- return new HiveTableOutputFormat(
- jobConf,
- tablePath,
- catalogTable,
- hiveTablePartition,
- HiveReflectionUtils.getTableMetadata(hiveShim,
table),
- overwrite);
+
+ FileSystemOutputFormat.Builder<Row> builder = new
FileSystemOutputFormat.Builder<>();
+ builder.setColumnNames(tableSchema.getFieldNames());
+
builder.setDefaultPartName(jobConf.get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+
HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal));
+ builder.setDynamicGrouped(dynamicGrouping);
+ builder.setPartitionColumns(partitionColumns);
+ builder.setFileSystemFactory(new
HiveFileSystemFactory(jobConf));
+ builder.setFormatFactory(new HiveOutputFormatFactory(
+ jobConf,
+ sd.getOutputFormat(),
+ sd.getSerdeInfo(),
+ tableSchema,
+ partitionColumns,
+
HiveReflectionUtils.getTableMetadata(hiveShim, table),
+ hiveVersion));
+ builder.setMetaStoreFactory(
+ new HiveMetaStoreFactory(jobConf,
hiveVersion, dbName, tableName));
+ builder.setOverwrite(overwrite);
+ builder.setStaticPartitions(staticPartitionSpec);
+ builder.setTmpPath(new org.apache.flink.core.fs.Path(
Review comment:
Note temp file is come from table location, `FileSystemOutputFormat` have no
idea to how to generate temp file. Actually, `FileSystemOutputFormat` just need
know temp dir, it don't need know final path.
Now the generation of the temp directory is uncertain. You can see below
hive codes have todo. I don't want bring this thing to `FileSystemOutputFormat`
now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
