EMsnap commented on code in PR #8096:
URL: https://github.com/apache/inlong/pull/8096#discussion_r1209085490
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/HiveTableSink.java:
##########
@@ -172,23 +199,50 @@ private DataStreamSink<?> consume(
try (HiveMetastoreClientWrapper client =
HiveMetastoreClientFactory.create(HiveConfUtils.create(jobConf), hiveVersion)) {
- Table table = client.getTable(dbName, identifier.getObjectName());
- StorageDescriptor sd = table.getSd();
-
- Class hiveOutputFormatClz =
-
hiveShim.getHiveOutputFormatClass(Class.forName(sd.getOutputFormat()));
+ StorageDescriptor sd;
+ Properties tableProps = new Properties();
+ Class hiveOutputFormatClz;
+ HiveWriterFactory writerFactory;
+ boolean sinkMultipleEnable = Boolean.parseBoolean(
+
catalogTable.getOptions().getOrDefault(SINK_MULTIPLE_ENABLE.key(), "false"));
boolean isCompressed =
jobConf.getBoolean(HiveConf.ConfVars.COMPRESSRESULT.varname, false);
- HiveWriterFactory writerFactory =
- new HiveWriterFactory(
- jobConf,
- hiveOutputFormatClz,
- sd.getSerdeInfo(),
- tableSchema,
- getPartitionKeyArray(),
- HiveReflectionUtils.getTableMetadata(hiveShim,
table),
- hiveShim,
- isCompressed);
+ if (sinkMultipleEnable) {
+ sd = new StorageDescriptor();
+ SerDeInfo serDeInfo = new SerDeInfo();
+
serDeInfo.setSerializationLib("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe");
+ sd.setSerdeInfo(serDeInfo);
+ String defaultFs = jobConf.get("fs.defaultFS", "");
+ sd.setLocation(defaultFs + "/tmp");
+ hiveOutputFormatClz =
hiveShim.getHiveOutputFormatClass(HiveIgnoreKeyTextOutputFormat.class);
+
+ writerFactory = new HiveWriterFactory(
Review Comment:
here new HiveWriterFactory( ) can be extracted out of the if statement,
which applies also in else statement
--
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-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
