How do we use query configurations while using SQL client in Flink SQL?

Question

The same fashion as mentioned in the link below for https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/query_configuration.html

Want to use Idle state retention time.

David Anderson · Accepted Answer · 2019-09-17 11:46:22Z

1

Flink's SQL client can read a YAML configuration file on startup, and this file can include idle state retention specifications. For example, you might start the client via

sql-client.sh embedded -e sql-client-config.yaml

where the config file contains

execution:
  planner: old                      # optional: either 'old' (default) or 'blink'
  type: streaming                   # required: execution mode either 'batch' or 'streaming'
  result-mode: table                # required: either 'table' or 'changelog'
  max-table-result-rows: 1000000    # optional: maximum number of maintained rows in
                                    #   'table' mode (1000000 by default, smaller 1 means unlimited)
  time-characteristic: event-time   # optional: 'processing-time' or 'event-time' (default)
  parallelism: 1                    # optional: Flink's parallelism (1 by default)
  periodic-watermarks-interval: 200 # optional: interval for periodic watermarks (200 ms by default)
  min-idle-state-retention: 0       # optional: table program's minimum idle state time
  max-idle-state-retention: 0       # optional: table program's maximum idle state time

See the docs for more details.

answered Sep 17, 2019 at 11:46

David Anderson

44.3k4 gold badges41 silver badges73 bronze badges

Sign up to request clarification or add additional context in comments.

6 Comments

Nishant Gupta Over a year ago

Thanks David for replying back quick. Does this min-idle-state-retention, max-idle-state-retention holds appropriate for query like SELECT sourceKafka.* FROM sourceKafka INNER JOIN badips ON sourceKafka.source.ip=badips.ip WHERE sourceKafka.timestamp_received BETWEEN CURRENT_TIMESTAMP - INTERVAL '15' MINUTE AND CURRENT_TIMESTAMP; Since there is no grouping and I read that these parameters define for how long the state of a key is retained

David Anderson Over a year ago

For a join like that, each stream will be keyed by the ip, so yes, this will be keyed state that is subject to the state retention policy. But moreover, you don't need to bother with this, because with a time windowed join, Flink SQL will automatically clear older records that have become irrelevant.

Nishant Gupta Over a year ago

Time windowed join, Flink SQL will automatically clear older records that have become irrelevant => That is what I also thought so... but when i tried the above query, 60+GB heap space runs out. Don't know why

David Anderson Over a year ago

That's potentially a rather complex topic, not well suited for stack overflow. Suggest you ask for help on the user mailing list.

Nishant Gupta Over a year ago

Sure David. Thanks

|

Collectives™ on Stack Overflow

How do we use query configurations while using SQL client in Flink SQL?

1 Answer 1

6 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

6 Comments

Your Answer

Sign up or log in

Post as a guest

Related