https://flink.apache.org/ logo
Join Slack
Powered by
Hi all! Is there any option to control partition k...
# troubleshooting
m
Hi all! Is there any option to control partition key in datastream api + kinesis sink + PyFlink? In Table API, i use PARTITIONED BY ( some_id ) and it works pretty well. In Datastream API, i see only to options when asetting up kinesis streaming sink - PartitionKeyGenerator.random() and PartitionKeyGenerator.fixed(), and i can’t find way to pass some part of item there. Is there such way?
a
I think you're looking for key_by which lets you specify a key selector rather than generator
m
i explored this option. Does this mean that PartitionKeyGenerator.fixed() + keyed stream will result in this keys be in kinesis sink partition key? Seems to me that this doesnt happen
Copy code
t_env.execute_sql(f"""
        CREATE TABLE data (
            o           STRING,
            price        INT
        ) WITH (
            'connector' = 'datagen',
            'rows-per-second' = '1'
        )
    """)

ds = t_env.to_data_stream(t_env.from_path("data"))
keyed = ds.key_by(lambda x: x[1], key_type=<http://Types.INT|Types.INT>())
keyed = keyed.map(lambda x: f"new - {x[0]} - {x[1]} - 11\n", output_type=Types.STRING())

keyed.sink_to(kds_sink)
env.execute()
this - doesn’t work. Key is still fixed “0”…
Found this in java docs A partitioner ensuring that each internal Flink partition ends up in the same Kinesis partition. This is achieved by using the index of the producer task as a 
PartitionKey
. https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/connector/kinesis/table/FixedKinesisPartitionKeyGenerator.html Seems to me it’s impossible, as python kinesis sink PartitionKeyGenerator has only two options, both of which are actually gateways to Java versions and no abliilty to have othewr ones….
a
I think you want a key selector, not generator
if you have a user/client/customer/session ID that you want to use as a group-by, so that aggregates and other stateful computations are local to that specific ID value, use a key selector that produces the identifier
e.g. if you have events {clientId: 123}, {clientId: 234}, {clientId: 345}, {clientId: 123} and your key selector returns the value of 
clientId
, then a 
count(*)
(assuming they all arrive within the window) will return: • 123: 2 • 234: 1 • 345: 1
Copy code
val events: DataStream[X] = ...
val keyed = events.keyBy(_.clientId)
4 Views
Open in Slack
PreviousNext
Powered by