Hi, Is there any way to partitioned data sink data...
# troubleshooting
s
Hi, Is there any way to partitioned data sink data in a file system?
Copy code
final StreamingFileSink<String> sink = StreamingFileSink
    .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
            .withMaxPartSize(1024 * 1024 * 1024)
            .build())
	.build();
The above code will assign records to the default one hour time buckets. I want to write all the parts in same directory, don't want hourly bucket. Is there is way to do so?
a
You could add a bucket assigner before the build and it should help you set the same directory. The path input you pass would be the base path and you can configure a static value in bucket assigner.
Copy code
StreamingFileSink<Tuple2<Integer, Integer>> sink = StreamingFileSink
 .forRowFormat((new Path(outputPath), new SimpleStringEncoder<>("UTF-8"))
 .withBucketAssigner(new KeyBucketAssigner())
 .withRollingPolicy(OnCheckpointRollingPolicy.build())
 .withOutputFileConfig(config)
 .build();
https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/streamfile_sink/#part-file-lifecycle
s
Thanks @Alokh P. I'll try.
It workedπŸ˜€
πŸ™Œ 1
Hi @Alokh P Do you also know, how to write a flink ssql in Zeppelin that will continuously update results that are in part? Right now I have to run paragraphs again and again to update the results present in the new part. For a particular part, it will update automatically, but as soon as a new part is created, I have to run the paragraph again.
a
Maybe try writing data in a different format like iceberg or delta etc. which supports updates
If I understood correctly updates is what you need and you are currently running a deduplication process asynchronously
s
Actually, Zeppelin supports Flink 1.13 and directory-watching is unsupported in Flink 1.13..
a
If you move to write data into Iceberg for example you wouldnt need to have a different process to update. The main write itself would take care of it. https://iceberg.apache.org/docs/latest/flink-writes/
s
Will give it a try πŸ’―