Hi Alexander, Routing of input data in Flink can be done through keying and this can guarantee collocation constraints. This means that you can send two records to the same node by giving them the same key, e.g. the topic name. Keep in mind that elements with different keys do not necessarily go to different nodes, as key assignment to nodes is random.
Given this, you could initially key by topic, so that all records of a topic go to the same node. This node will compute statistics about the topic, e.g. elem/sec (t) and based on thresholds assign new keys to each record, e.g. TOPIC-1 if t < 1000, TOPIC-2 if t >= 1000 && t < 2000, etc and re-key. This will not guarantee that TOPIC-1 and TOPIC-2 will go to different machines but the probability of this happening will increase with the parallelism of your job. Finally, based on your bucket assigner and the rolling policy, you can redirect the elements to the same bucket, e.g. TOPIC and tune how many part-files you will have based on part-file size and/or time. Will this help you with your use-case? Cheers, Kostas On Thu, Jun 25, 2020 at 3:23 AM Alexander Filipchik <afilipc...@gmail.com> wrote: > > Maybe I misreading the documentation, but: > "Data within the partition directories are split into part files. Each > partition will contain at least one part file for each subtask of the sink > that has received data for that partition." > > So, it is 1 partition per subtask. I'm trying to figure out how to > dynamically adjust which subtask is getting the data to minimize the number > of subtasks writing into a specific partition. > > Alex > > On Wed, Jun 24, 2020 at 3:55 PM Seth Wiesman <sjwies...@gmail.com> wrote: >> >> You can achieve this in Flink 1.10 using the StreamingFileSink. >> >> I’d also like to note that Flink 1.11 (which is currently going through >> release testing and should be available imminently) has support for exactly >> this functionality in the table API. >> >> https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/filesystem.html >> >> >> On Wed, Jun 24, 2020 at 1:53 PM Alexander Filipchik <afilipc...@gmail.com> >> wrote: >>> >>> Hello! >>> >>> We are working an a Flink Streaming job that reads data from multiple Kafka >>> topics and writes them to DFS. We are using StreamingFileSink with custom >>> implementation for GCS FS and it generates a lot of files as streams are >>> partitioned among multiple JMs. In the ideal case we should have at most 1 >>> file per kafka topic per interval. We also have some heavy topics and some >>> pretty light ones, so the solution should also be smart to utilize >>> resources efficiently. >>> >>> I was thinking we can partition based on how much data is ingested in the >>> last minute or so to make sure: messages from the same topic are routed to >>> the same (or minimal number of ) file if there are enough resources to do >>> so. Think bin packing. >>> >>> Is it a good idea? Is there a built in way to achieve it? If not, is there >>> a way to push state into the partitioner (or even kafka client to >>> repartition in the source)? I was thinking that I can run a side stream >>> that will calculate data volumes and then broadcast it into the main >>> stream, so partitioner can make a decision, but it feels a bit complex. >>> >>> Another way is to modify kafka client to track messages per topics and make >>> decision at that layer. >>> >>> Am I on the right path? >>> >>> Thank you
