Skip to content

Commit f54d2ec

Browse files
committed
解决kudu 结果表没有Metric问题
1 parent b9f115f commit f54d2ec

File tree

1 file changed

+6
-1
lines changed
  • kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu

1 file changed

+6
-1
lines changed

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import org.apache.flink.api.java.typeutils.RowTypeInfo;
1010
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
1111
import org.apache.flink.streaming.api.datastream.DataStream;
12+
import org.apache.flink.streaming.api.datastream.DataStreamSink;
1213
import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
1314
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
1415
import org.apache.flink.table.sinks.RetractStreamTableSink;
@@ -63,7 +64,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
6364
.setFieldTypes(this.fieldTypes);
6465
KuduOutputFormat kuduOutputFormat = builder.finish();
6566
RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(kuduOutputFormat);
66-
dataStream.addSink(richSinkFunction);
67+
DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
68+
dataStreamSink.name(tableName);
69+
if (parallelism > 0) {
70+
dataStreamSink.setParallelism(parallelism);
71+
}
6772
}
6873

6974
@Override

0 commit comments

Comments
 (0)