ChaomingZhangCN commented on code in PR #666:
URL: https://github.com/apache/geaflow/pull/666#discussion_r2517078455
##########
geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-odps/src/main/java/org/apache/geaflow/dsl/connector/odps/OdpsTableSink.java:
##########
@@ -89,54 +86,41 @@ public void init(Configuration tableConf, StructType
schema) {
this.tableName =
tableConf.getString(OdpsConfigKeys.GEAFLOW_DSL_ODPS_TABLE);
this.accessKey =
tableConf.getString(OdpsConfigKeys.GEAFLOW_DSL_ODPS_ACCESS_KEY);
this.accessId =
tableConf.getString(OdpsConfigKeys.GEAFLOW_DSL_ODPS_ACCESS_ID);
- this.partitionSpec =
tableConf.getString(OdpsConfigKeys.GEAFLOW_DSL_ODPS_PARTITION_SPEC);
- this.shardNamePrefix = project + "-" + tableName + "-";
int bufferSize =
tableConf.getInteger(OdpsConfigKeys.GEAFLOW_DSL_ODPS_SINK_BUFFER_SIZE);
if (bufferSize > 0) {
this.bufferSize = bufferSize;
}
+ int flushIntervalMs =
tableConf.getInteger(OdpsConfigKeys.GEAFLOW_DSL_ODPS_SINK_FLUSH_INTERVAL_MS);
+ if (flushIntervalMs > 0) {
+ this.flushIntervalMs = flushIntervalMs;
+ }
int timeoutSeconds =
tableConf.getInteger(OdpsConfigKeys.GEAFLOW_DSL_ODPS_TIMEOUT_SECONDS);
if (timeoutSeconds > 0) {
this.timeoutSeconds = timeoutSeconds;
}
checkArguments();
- LOGGER.info("init odps table sink, endPoint : {}, project : {},
tableName : {}, "
- + "partitionSpec: {}", endPoint, project, tableName,
partitionSpec);
+ LOGGER.info("init odps table sink, endPoint : {}, project : {},
tableName : {}",
+ endPoint, project, tableName);
}
@Override
public void open(RuntimeContext context) {
- this.buffer = new ArrayList<>(bufferSize);
Account account = new AliyunAccount(accessId, accessKey);
- this.odps = new Odps(account);
+ Odps odps = new Odps(account);
odps.setEndpoint(endPoint);
odps.setDefaultProject(project);
- this.table = odps.tables().get(project, tableName);
this.tunnel = new TableTunnel(odps);
- if (StringUtils.isEmpty(partitionSpec)) {
- throw new GeaFlowDSLException("For ODPS sink, partition spec
cannot be empty.");
- }
- PartitionSpec usePartitionSpec = new PartitionSpec(partitionSpec);
- ExecutorService executor = Executors.newSingleThreadExecutor();
- Future<UploadSession> future = executor.submit(() -> {
- try {
- return tunnel.createUploadSession(project, tableName,
usePartitionSpec);
- } catch (TunnelException e) {
- throw new GeaFlowDSLException("Cannot get odps session.", e);
- }
- });
- try {
- this.uploadSession = future.get(this.timeoutSeconds,
TimeUnit.SECONDS);
- } catch (Exception e) {
- throw new GeaFlowDSLException("Cannot list partitions from ODPS,
endPoint: " + this.endPoint, e);
- }
- this.recordColumns =
this.uploadSession.getSchema().getColumns().toArray(new Column[0]);
+ TableSchema tableSchema = createUploadSession().getSchema();
Review Comment:
StreamUploadSession dose have close() method.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
