sql-client-conf.yaml

################################################################################
# Copyright 2019 Ververica GmbH
#
# Licensed 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
#
#  http://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.
################################################################################


# This file defines the default environment for Flink's SQL Client.
# Defaults might be overwritten by a session specific environment.


#==============================================================================
# Table Sources
#==============================================================================

# Define table sources here. See the Table API & SQL documentation for details.

tables:
  - name: D1_CARS
    type: source
    update-mode: append
    schema:
    - name: user
      type: STRING
    - name: rid
      type: BIGINT
    - name: manufacturer
      type: STRING
    - name: model
      type: STRING
    - name: plate
      type: STRING
    - name: status
      type: STRING
    - name: ts
      type: STRING
    - name: parsed_timestamp
      type: TIMESTAMP
      rowtime:
        timestamps:
          type: "from-field"
          from: "ts"
        watermarks:
          type: "periodic-bounded"
          delay: "5"
    connector:
      property-version: 1
      type: kafka
      version: universal
      topic: DEALER1_CARS
      startup-mode: earliest-offset
      properties:
      - key: bootstrap.servers
        value: kafka:9092
    format:
      property-version: 1
      type: json
      schema: "ROW(user STRING, rid BIGINT, manufacturer STRING, model STRING, plate STRING, status STRING, ts STRING)"
      
  - name: D1_TRUCKS
    type: source
    update-mode: append
    schema:
    - name: user
      type: STRING
    - name: rid
      type: BIGINT
    - name: manufacturer
      type: STRING
    - name: model
      type: STRING
    - name: plate
      type: STRING
    - name: status
      type: STRING
    - name: ts
      type: STRING
    - name: parsed_timestamp
      type: TIMESTAMP
      rowtime:
        timestamps:
          type: "from-field"
          from: "ts"
        watermarks:
          type: "periodic-bounded"
          delay: "5"
    connector:
      property-version: 1
      type: kafka
      version: universal
      topic: DEALER1_TRUCKS
      startup-mode: earliest-offset
      properties:
      - key: bootstrap.servers
        value: kafka:9092
    format:
      property-version: 1
      type: json
      schema: "ROW(user STRING, rid BIGINT, manufacturer STRING, model STRING, plate STRING, status STRING, ts STRING)"

  - name: D2_VEHICLES
    type: source
    update-mode: append
    schema:
    - name: userID
      type: BIGINT
    - name: rid
      type: BIGINT
    - name: type
      type: STRING
    - name: manufacturer
      type: STRING
    - name: model
      type: STRING
    - name: plate
      type: STRING
    - name: status
      type: STRING
    - name: ts
      type: STRING
    - name: parsed_timestamp
      type: TIMESTAMP
      rowtime:
        timestamps:
          type: "from-field"
          from: "ts"
        watermarks:
          type: "periodic-bounded"
          delay: "5"
    connector:
      property-version: 1
      type: kafka
      version: universal
      topic: DEALER2_VEHICLES
      startup-mode: earliest-offset
      properties:
      - key: bootstrap.servers
        value: kafka:9092
    format:
      property-version: 1
      type: json
      schema: "ROW(userID BIGINT, rid BIGINT, type STRING, manufacturer STRING, model STRING, plate STRING, status STRING, ts STRING)"

  - name: D2_USERS
    type: source
    update-mode: append
    schema:
    - name: userID
      type: BIGINT
    - name: name
      type: STRING
    connector:
      property-version: 1
      type: kafka
      version: universal
      topic: DEALER2_USERS
      startup-mode: earliest-offset
      properties:
      - key: bootstrap.servers
        value: kafka:9092
    format:
      property-version: 1
      type: json
      schema: "ROW(userID BIGINT, name STRING)"


#==============================================================================
# Execution properties
#==============================================================================

# Execution properties allow for changing the behavior of a table program.

execution:
  planner: blink               # using the Blink planner
  type: streaming              # 'batch' or 'streaming' execution
  result-mode: table           # 'changelog' or 'table' presentation of results
  parallelism: 1               # parallelism of the program
  max-parallelism: 128         # maximum parallelism
  min-idle-state-retention: 0  # minimum idle state retention in ms
  max-idle-state-retention: 0  # maximum idle state retention in ms

#==============================================================================
# Execution properties
#==============================================================================

# Flink configuration parameters

configuration:
  execution.checkpointing.interval: 1s

#==============================================================================
# Deployment properties
#==============================================================================

# Deployment properties allow for describing the cluster to which table
# programs are submitted to.

deployment:
  type: standalone             # only the 'standalone' deployment is supported
  response-timeout: 5000       # general cluster communication timeout in ms
  gateway-address: ""          # (optional) address from cluster to gateway
  gateway-port: 0              # (optional) port from cluster to gateway