(DEPRECATED) Apache Flink User Mailing List archive.

Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

classic Classic list List threaded Threaded
♦
♦
 8 messages Options
Mich Talebzadeh
Reply | Threaded
Open this post in threaded view
♦
♦
|

Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Mich Talebzadeh
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Will Du
Reply | Threaded
Open this post in threaded view
♦
♦
|

Kafka Avro Table Source

Will Du
Hi folks,
I am working on using avro table source mapping to kafka source. By looking at the example, I think the current Flink v1.5.0 connector is not flexible enough. I wonder if I have to specify the avro record class to read from Kafka.

For withSchema, the schema can get from schema registry. However, the class of avro seems hard coded.

thanks,
Will

KafkaTableSource source = Kafka010AvroTableSource.builder()
  // set Kafka topic
  .forTopic("sensors")
  // set Kafka consumer properties
  .withKafkaProperties(kafkaProps)
  // set Table schema
  .withSchema(TableSchema.builder()
    .field("sensorId", Types.LONG())
    .field("temp", Types.DOUBLE())
    .field("time", Types.SQL_TIMESTAMP()).build())
  // set class of Avro record
  .forAvroRecordClass(SensorReading.class)  // ? Any way to get this without hard code class
  .build();
Hequn Cheng
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Hequn Cheng
In reply to this post by Mich Talebzadeh
Hi Mich, 

It seems the writeMode has not been set correctly. Have you ever tried 
.writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE);

On Mon, Jul 2, 2018 at 10:44 PM, Mich Talebzadeh <[hidden email]> wrote:
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Mich Talebzadeh
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Mich Talebzadeh
thanks Hequn.

When I use as suggested, I am getting this error

error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:30: not found: value FileSystem
[error]                  .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
[error]                                                        ^
[error] one error found

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 03:16, Hequn Cheng <[hidden email]> wrote:
Hi Mich, 

It seems the writeMode has not been set correctly. Have you ever tried 
.writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE);

On Mon, Jul 2, 2018 at 10:44 PM, Mich Talebzadeh <[hidden email]> wrote:
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Jörn Franke
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Jörn Franke 
import org.apache.flink.core.fs.FileSystem


On 3. Jul 2018, at 08:13, Mich Talebzadeh <[hidden email]> wrote:

thanks Hequn.

When I use as suggested, I am getting this error

error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:30: not found: value FileSystem
[error]                  .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
[error]                                                        ^
[error] one error found

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 03:16, Hequn Cheng <[hidden email]> wrote:
Hi Mich, 

It seems the writeMode has not been set correctly. Have you ever tried 
.writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE);

On Mon, Jul 2, 2018 at 10:44 PM, Mich Talebzadeh <[hidden email]> wrote:
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Mich Talebzadeh
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Mich Talebzadeh
thanks Hequn and Jorn that helped.

But I am still getting this error for a simple streaming program at execution!

import java.util.Properties
import java.util.Arrays
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.streaming.util.serialization.DeserializationSchema
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.flink.core.fs.FileSystem

object md_streaming
{
  def main(args: Array[String])
  {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "rhes75:9092")
    //properties.setProperty("zookeeper.connect", "rhes75:2181")
    properties.setProperty("group.id", "md_streaming")
    properties.setProperty("auto.offset.reset", "latest")
    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
    env.execute("Flink Kafka Example")
  }
}


Completed compiling
Starting execution of program
------------------------------------------------------------
 The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign(Ljava/util/List;)V
        at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:265)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:464)
        at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)
        at md_streaming$.main(md_streaming.scala:32)
        at md_streaming.main(md_streaming.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:528)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:420)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:404)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:781)
        at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:275)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210)
        at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020)
        at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
        at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096)
Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign(Ljava/util/List;)V
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerCallBridge.assignPartitions(KafkaConsumerCallBridge.java:42)
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.reassignPartitions(KafkaConsumerThread.java:405)
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:243)



Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 07:20, Jörn Franke <[hidden email]> wrote:
import org.apache.flink.core.fs.FileSystem


On 3. Jul 2018, at 08:13, Mich Talebzadeh <[hidden email]> wrote:

thanks Hequn.

When I use as suggested, I am getting this error

error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:30: not found: value FileSystem
[error]                  .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
[error]                                                        ^
[error] one error found

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 03:16, Hequn Cheng <[hidden email]> wrote:
Hi Mich, 

It seems the writeMode has not been set correctly. Have you ever tried 
.writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE);

On Mon, Jul 2, 2018 at 10:44 PM, Mich Talebzadeh <[hidden email]> wrote:
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Chesnay Schepler
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Chesnay Schepler
This issue is covered in your other ML thread "java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign".

Let's move further discussions there so we don't have 2 threads in parallel for the same problem.

On 03.07.2018 09:21, Mich Talebzadeh wrote:
thanks Hequn and Jorn that helped.

But I am still getting this error for a simple streaming program at execution!

import java.util.Properties
import java.util.Arrays
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.streaming.util.serialization.DeserializationSchema
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.flink.core.fs.FileSystem

object md_streaming
{
  def main(args: Array[String])
  {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "rhes75:9092")
    //properties.setProperty("zookeeper.connect", "rhes75:2181")
    properties.setProperty("group.id", "md_streaming")
    properties.setProperty("auto.offset.reset", "latest")
    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
    env.execute("Flink Kafka Example")
  }
}


Completed compiling
Starting execution of program
------------------------------------------------------------
 The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign(Ljava/util/List;)V
        at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:265)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:464)
        at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)
        at md_streaming$.main(md_streaming.scala:32)
        at md_streaming.main(md_streaming.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:528)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:420)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:404)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:781)
        at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:275)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210)
        at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1020)
        at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1096)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
        at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1096)
Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign(Ljava/util/List;)V
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerCallBridge.assignPartitions(KafkaConsumerCallBridge.java:42)
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.reassignPartitions(KafkaConsumerThread.java:405)
        at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:243)



Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 07:20, Jörn Franke <[hidden email]> wrote:
import org.apache.flink.core.fs.FileSystem

On 3. Jul 2018, at 08:13, Mich Talebzadeh <[hidden email]> wrote:

thanks Hequn.

When I use as suggested, I am getting this error

error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:30: not found: value FileSystem
[error]                  .writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE)
[error]                                                        ^
[error] one error found

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Tue, 3 Jul 2018 at 03:16, Hequn Cheng <[hidden email]> wrote:
Hi Mich, 

It seems the writeMode has not been set correctly. Have you ever tried 
.writeAsText("/tmp/md_streaming.txt", FileSystem.WriteMode.OVERWRITE);

On Mon, Jul 2, 2018 at 10:44 PM, Mich Talebzadeh <[hidden email]> wrote:
Flink 1.5

This streaming data written to a file

    val stream = env
                 .addSource(new FlinkKafkaConsumer09[String]("md", new SimpleStringSchema(), properties))
                 .writeAsText("/tmp/md_streaming.txt")
 env.execute("Flink Kafka Example")

The error states

Caused by: java.io.IOException: File or directory /tmp/md_streaming.txt already exists. Existing files and directories are not overwritten in NO_OVERWRITE mode. Use OVERWRITE mode to overwrite existing files and directories.

Is there any append in writeAsText? I tried OVERWRITE but  did not work.

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 
Fabian Hueske-2
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Kafka Avro Table Source

Fabian Hueske-2
In reply to this post by Will Du
Hi Will,

The community is currently working on improving the Kafka Avro integration for Flink SQL.
There's a PR [1]. If you like, you could try it out and give some feedback.

Timo (in CC) has been working Kafka Avro and should be able to help with any specific questions.

Best, Fabian


2018-07-03 3:02 GMT+02:00 Will Du <[hidden email]>:
Hi folks,
I am working on using avro table source mapping to kafka source. By looking at the example, I think the current Flink v1.5.0 connector is not flexible enough. I wonder if I have to specify the avro record class to read from Kafka.

For withSchema, the schema can get from schema registry. However, the class of avro seems hard coded.

thanks,
Will

KafkaTableSource source = Kafka010AvroTableSource.builder()
  // set Kafka topic
  .forTopic("sensors")
  // set Kafka consumer properties
  .withKafkaProperties(kafkaProps)
  // set Table schema
  .withSchema(TableSchema.builder()
    .field("sensorId", Types.LONG())
    .field("temp", Types.DOUBLE())
    .field("time", Types.SQL_TIMESTAMP()).build())
  // set class of Avro record
  .forAvroRecordClass(SensorReading.class)  // ? Any way to get this without hard code class
  .build();
Free forum by Nabble Edit this page