quickstart-flink/quickstart-flinksql/flinksql_1.14/assembly/conf/application.yml

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
flink:
  option:
    target: yarn-per-job
    detached:
    shutdownOnAttachedExit:
    zookeeperNamespace:
    jobmanager:
  property: #@see: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/
    $internal.application.main: org.apache.streampark.quickstart.flink.flinksql.JavaFlinkSqlJob
    pipeline.name: streampark quickstart flinksql job
    taskmanager.numberOfTaskSlots: 1
    parallelism.default: 2
    jobmanager.memory:
      flink.size:
      heap.size:
      jvm-metaspace.size:
      jvm-overhead.max:
      off-heap.size:
      process.size:
    taskmanager.memory:
      flink.size:
      framework.heap.size:
      framework.off-heap.size:
      managed.size:
      process.size:
      task.heap.size:
      task.off-heap.size:
      jvm-metaspace.size:
      jvm-overhead.max:
      jvm-overhead.min:
      managed.fraction: 0.4
    pipeline:
      auto-watermark-interval: 200ms
    # checkpoint
    execution:
      checkpointing:
        mode: EXACTLY_ONCE
        interval: 30s
        timeout: 10min
        unaligned: false
        externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
    # state backend
    state:
      backend: hashmap # Special note: flink1.12 optional configuration ('jobmanager', 'filesystem', 'rocksdb'), flink1.12+ optional configuration ('hashmap', 'rocksdb'),
      backend.incremental: true
      checkpoint-storage: filesystem
      savepoints.dir: file:///tmp/chkdir
      checkpoints.dir: file:///tmp/chkdir
    # restart strategy
    restart-strategy: fixed-delay  # Restart strategy [(fixed-delay|failure-rate|none) a total of 3 configurable strategies]
    restart-strategy.fixed-delay:
      attempts: 3
      delay: 5000
    restart-strategy.failure-rate:
      max-failures-per-interval:
      failure-rate-interval:
      delay:
  # table
  table:
    table.local-time-zone: default # @see https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config/


sql:
  flinksql: |
    CREATE TABLE datagen (
      f_sequence INT,
      f_random INT,
      f_random_str STRING,
      ts AS localtimestamp,
      WATERMARK FOR ts AS ts
    ) WITH (
      'connector' = 'datagen',
      -- optional options --
      'rows-per-second'='5',
      'fields.f_sequence.kind'='sequence',
      'fields.f_sequence.start'='1',
      'fields.f_sequence.end'='1000',
      'fields.f_random.min'='1',
      'fields.f_random.max'='1000000',
      'fields.f_random_str.length'='10'
    );

    CREATE TABLE print_table (
      f_sequence INT,
      f_random INT,
      f_random_str STRING
      ) WITH (
      'connector' = 'print'
    );

    INSERT INTO print_table select f_sequence,f_random,f_random_str from datagen;