Environment Setup¶

• Create a Python virtual environment:

  python -m venv .venv
  

• Activate the environment:

  source .venv/bin/activate
  

• Be sure to use the pip install in the virtual environment:

  python -m pip --version
  # if pip not present
  python -m ensurepip
  # install requirements
  python -m pip install -r requirements.txt
  

• Install the shift_left CLI using the command (this is temporary once the CLI will be loaded to pypi): To get the list of version of the wheel, they are under the src/shift_left/dist folder

  pip install src/shift_left/dist/shift_left-0.1.30-py3-none-any.whl
  

• Create a new virtual environment in any folder, but could be the : uv venv
• Activate the environment:

  source .venv/bin/activate
  

• Install the cli:

  uv tool list
  # in case it was already installed
  uv uninstall shift_left
  uv install shift_left src/shift_left/dist/shift_left-0.1.30-py3-none-any.whl
  

• Get the Kakfa cluster URL using the Confluent Console > Cluster Settings page. The URL has the cluster id and a second id that is used for the RESP API.

kafka:
  bootstrap.servers: lkc-2qxxxx-pyyyy.us-west-2.aws.confluent.cloud:9092
  cluster_id: lkc-2qxxxx
  pkafka_cluster: lkc-2qxxxx-pyyyy
  security.protocol: SASL_SSL
  sasl.mechanisms: PLAIN
  sasl.username: <key name>
  sasl.password: <key seceret> 
  session.timeout.ms: 5000
  cluster_type: stage
  src_topic_prefix: clone

• Get environment ID from the Environment details in the Confluen Console. The cloud provider and region.

confluent_cloud:
  base_api: api.confluent.cloud/org/v2
  environment_id:  env-20xxxx
  region:  us-west-2
  provider:  aws
  organization_id: 5329.....96
  api_key: T3.....HH
  api_secret: secret-key
  page_size: 100
  glb_name: glb
  url_scope: private

• Flink settings are per environment. Get the URL endpoint by going to Environments > one_of_the_env > Flink > Endpoints, copy the private or public endpoint

  flink:
      flink_url: flink-d....7.us-west-2.aws.glb.confluent.cloud
      api_key: MVXI.....HY   
      api_secret: cf.......IztA
      compute_pool_id: lfcp-0...
      catalog_name:  dev-flink-us-west-2
      database_name:  dev-flink-us-west-2
      max_cfu: 50       #-- maximum CFU when creating compute pool via the tool
      max_cfu_percent_before_allocation: .8   #-- percent * 100 to consider before creating a new compute pool 
  

• Define the API keys from Environments > one_of_the_env > Flink > API keys.

• The compute pool id is used as default for running Flink query.
• Catalog name is the name of the environment and database name is the name of the Kafka cluster

The app section defines a set of capabilities to tune the cli.

app:
    delta_max_time_in_min: 15  # this is to apply randomly to the event timestamp to create out of order
    report_output_dir: ./reports
    default_PK: __db
    timezone: America/Los_Angeles
    src_table_name_prefix: src_
    logging: INFO
    products: ["p1", "p2", "p3"]
    accepted_common_products: ['common', 'seeds']
    cache_ttl: 120 # in seconds
    sql_content_modifier: shift_left.core.utils.table_worker.ReplaceEnvInSqlContent
    translator_to_flink_sql_agent: shift_left.core.utils.translator_to_flink_sql.DbtTranslatorToFlinkSqlAgent
    dml_naming_convention_modifier: shift_left.core.utils.naming_convention.DmlNameModifier
    compute_pool_naming_convention_modifier: shift_left.core.utils.naming_convention.ComputePoolNameModifier
    data_limit_where_condition : rf"where tenant_id in ( SELECT tenant_id FROM tenant_filter_pipeline WHERE product = {product_name})"
    data_limit_replace_from_reg_ex: r"\s*select\s+\*\s+from\s+final\s*;?"
    data_limit_table_type: source
    data_limit_column_name_to_select_from: tenant_id
    post_fix_unit_test: _ut
    post_fix_integration_test: _it

• post_fix_unit_test, post_fix_integration_test are used to append the given string to table name during unit testing and integration test respectively.
• The data_limit_replace_from_reg_ex, data_limit_table_type, data_limit_column_name_to_select_from are used to add data filtering to all the source tables based on one column name to filter. The regex specifies to file the select * from final which is the last string in most Flink statements using CTEs implementation.
• sql_content_modifier specifies the custom class to use to do some SQL content modification depending of the target environment. This is a way to extend the CLI logic to specific usage.