Hey! I’m using the Flink Delta Connector

0.6.0

along with Flink

1.15.3

and things run super smoothly, thanks for the release! I have however noticed metric name collisions from the connector:

Name collision: Group already contains a Metric with the name 'DeltaSinkBytesWritten'. Metric will not be reported.[my_application_name, flink, operator]

A colleague of mine sent me this thread https://lists.apache.org/thread/xl02b08n823ct9lsnhosv2x2zlskm0cs, indicating that

To prevent this warning from occurring you will have to give every operator a unique name.

. The Flink application I have set up is quite simple adheres to the above:

val source = streamExecutionEnvironment
        .fromSource(
            KafkaRecordSource.create(configuration.sourceConfiguration, configuration.kafkaConfiguration),
            WatermarkStrategy.forMonotonousTimestamps(),
            configuration.sourceConfiguration.kafkaTopic
        )
        .flatMap(KafkaConsumerRecordRowMapper())
        .name("Kafka Consumer Record Row Mapper")

...

    source.sinkTo(sink).name("Delta Table Sink")
...
    streamExecutionEnvironment.execute("[application name] Kafka to Delta Lake")

A note is that I have prefixed the operator metrics scope:

...
metrics.scope.operator: my_application_name.flink.operator

The source thread that the log comes from is

Source: [redacted-kafka-topic-name] -> Kafka Consumer Record Row Mapper -> Delta Table Sink: Writer -> Delta Table Sink: Committer (3/3)#0

indicating that it’s not the Global Committer (to my understanding). The Kafka topic has three partitions, hence the source having three subtasks in the operator on Flink (see screenshot). Is this a bug or is there something I may have missed? 🙂

Hi, I am creating a Flink Delta sink pipleine where i am using Flink 16 and delta connector 0.6.0. From this https://issues.apache.org/jira/browse/FLINK-29856https://issues.apache.org/jira/browse/FLINK-29856, what i can understand is savepoint will behave different than checkpoint to not commit any side effect. I have seen that parquet files are getting closed on checkpoints but not on savepoints. I believe FLINK-29856 is the reason for that. My question is: why i see a new file getting opened on savepoint when file is not getting closed/committed. Can we clearly say when a new file gets opened, closed and committed? Also, the partitioning is happening at hour level, so data is not coming for the new partition after savepoint. Attaching the screenshot to show the savepoint and checkpoint timings. s3 ls is presenting file's opening time as well.

Hello, I want to understand the reason behind the different behaviour of Flink-delta pipelines. When i am running a Kafka in and delta out pipeline with fewer resources than needed for a particular pipeline, It fails with OOM error. In the newer version it takes around 8 minutes to give the first exception while in the older version it fails in 2 minutes only. what is the new change which can explain this behaviour? we have enabled the retrials, hence using the 'first' term. Pipeline1 Flink version: 1.16.1 Delta connector version: 0.6.0 Time taken for the first OOM error: 8 mins Pipeline2 Flink version: 1.13.6 Delta connector version: 0.5.0 Time taken for the first OOM error: 2 mins Attaching the screenshot of the error.

HI! Why the flink-connector requires hadoop-client in a project's compile scope? Couldn't it use the current mechanism in Flink to access hadoop stuff without having to have it in my application classpath? Thanks!

<!-- Needed for Flink Table/SQL API -->
      <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>${hadoop-version}</version>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-table-runtime_${scala.main.version}</artifactId>
          <version>${flink-version}</version>
          <scope>provided</scope>
      </dependency>
      <!-- End needed for Flink Table/SQL API -->

      <!-- Needed for AWS S3 support -->
      <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-aws</artifactId>
        <version>${hadoop-version}</version>
      </dependency>
      <!-- End needed for AWS S3 support -->

Hi? Anyone aware of this bug https://github.com/delta-io/delta/issues/1971

Hi, is it possible to consume data from a stream and write to multiple delta tables depending on a field in the event itself?