Hi Team I have been trying to `load local avro files` conver Apache Flink #troubleshooting

Hi Team, I have been trying to `load local avro fi...

Kaiwen Yuan

08/14/2023, 10:24 PM

Hi Team, I have been trying to

load local avro files

, convert them into source / stream to process but in vain. I have tried

FileSource.forRecordStreamFormat

but no luck. Does anyone happen to solve a problem like this before ?

💯 1

Benjamin Leff

08/14/2023, 11:07 PM

@Kaiwen Yuan I am looking for something similiar but for BulkFiles and parquet

Kaiwen Yuan

08/14/2023, 11:12 PM

ok, after some efforts, I kinda got it work. Basically, you will need to load avro files as dataset in one env, do collect; then in a stream env, you connect with "fromCollection".

Benjamin Leff

08/14/2023, 11:12 PM

any chance you can post your code? 🙏

Benjamin Leff

08/14/2023, 11:14 PM

my current code looks something like the following:

Copy code

val format =  ParquetColumnarRowInputFormat.createPartitionedFormat(
        hadoopConfig=new Configuration(),
        producedRowType=rowType,
        producedTypeInfo=rowTypeInfo,
        partitionKeys=partitionKeys,
        extractor=extractor,
        batchSize=2048,
        isUtcTimestamp=true,
        isCaseSensitive=false
      )

    // // Monitor the Paths to read data as unbounded data
    val source = FileSource
      .forBulkFileFormat(format, datalakeS3Path)
      .monitorContinuously(Duration.ofMillis(5L))
      .build()

    val dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "DataLake")

    dataStream

Kaiwen Yuan

08/14/2023, 11:24 PM

Copy code

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        AvroInputFormat<User> users = new AvroInputFormat<User>(new Path(avroFilePath), User.class);
        DataSet<User> userDataSet = env.createInput(users); 
        List<User> userDataList = userDataSet.collect();

        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<User> stream = streamEnv.fromCollection(userDataList);

Kaiwen Yuan

08/14/2023, 11:24 PM

I tried the filesource api but in vain

Kaiwen Yuan

08/14/2023, 11:25 PM

You basically want to treat them as DataSet, load them into memory, then use a stream env to load it out to connect the rest

Benjamin Leff

08/14/2023, 11:29 PM

@Kaiwen Yuan thank you! frustrating that

FileSource

api is not easy to use

Kaiwen Yuan

08/14/2023, 11:30 PM

yeah, I wasted 2 days on it 😞

Benjamin Leff

08/14/2023, 11:30 PM

same!

Simon Gabet

09/06/2023, 3:34 PM

hi @Kaiwen Yuan and @Benjamin Leff, did you had any success concerning the FileSource ? I am also trying to read avro files, but the

AvroInputFormat

is for the old API. I saw what you did but I don't want (nor can...) collect my dataset, so your solution would not work for me 😕

Benjamin Leff

09/06/2023, 6:21 PM

unfortunately i did not. due to a few reasons, we moved away from flink for this use case

Open in Slack

Previous Next