I try to run extract job in airflow got an error

[zemelLeong]

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 175, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 192, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/amundsen/db_metadata_extractor.py", line 73, in create_mysql_table_extract_job
    job.launch()
  File "/home/airflow/.local/lib/python3.7/site-packages/databuilder/job/job.py", line 76, in launch
    raise e
  File "/home/airflow/.local/lib/python3.7/site-packages/databuilder/job/job.py", line 64, in launch
    self._init()
  File "/home/airflow/.local/lib/python3.7/site-packages/databuilder/job/job.py", line 51, in _init
    self.task.init(self.conf)
  File "/home/airflow/.local/lib/python3.7/site-packages/databuilder/task/task.py", line 45, in init
    self.extractor.init(Scoped.get_scoped_conf(conf, self.extractor.get_scope()))
  File "/home/airflow/.local/lib/python3.7/site-packages/databuilder/extractor/sql_alchemy_extractor.py", line 29, in init
    self.conn_string = conf.get_string(SQLAlchemyExtractor.CONN_STRING)
  File "/home/airflow/.local/lib/python3.7/site-packages/pyhocon/config_tree.py", line 248, in get_string
    value = self.get(key, default)
  File "/home/airflow/.local/lib/python3.7/site-packages/pyhocon/config_tree.py", line 236, in get
    return self._get(ConfigTree.parse_key(key), 0, default)
  File "/home/airflow/.local/lib/python3.7/site-packages/pyhocon/config_tree.py", line 177, in _get
    u"No configuration setting found for key {key}".format(key='.'.join(key_path[:key_index + 1])))
pyhocon.exceptions.ConfigMissingException: 'No configuration setting found for key conn_string'

this is my config

def doris_conn_string():
    return f'mysql+pymysql://root:{urlquote("yxin-bd-dev@2023?!$")}@10.0.0.10:9030/db_resource'


def create_mysql_table_extract_job():
    exclude_schemas = ('information_schema')
    where_clause_suffix = f'where c.table_schema not in {exclude_schemas}'
    job_publish_tag = datetime.now().strftime('%Y-%m-%d')

    job_config = ConfigFactory.from_dict({
        f'extractor.mysql_metadata.{MysqlMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY}': where_clause_suffix,
        f'extractor.mysql_metadata.{MysqlMetadataExtractor.USE_CATALOG_AS_CLUSTER_NAME}': True,
        f'extractor.mysql_metadata.extractor.sqlalchemy.{SQLAlchemyExtractor.CONN_STRING}': doris_conn_string(),

        f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': csv_dir,
        f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': csv_dir,

        f'publisher.neo4j.{neo4j_csv_publisher.NODE_FILES_DIR}': csv_dir,
        f'publisher.neo4j.{neo4j_csv_publisher.RELATION_FILES_DIR}': csv_dir,
        f'publisher.neo4j.{neo4j_csv_publisher.NEO4J_END_POINT_KEY}': neo4j_endpoint,
        f'publisher.neo4j.{neo4j_csv_publisher.NEO4J_USER}': neo4j_user,
        f'publisher.neo4j.{neo4j_csv_publisher.NEO4J_PASSWORD}': neo4j_password,
        f'publisher.neo4j.{neo4j_csv_publisher.JOB_PUBLISH_TAG}': job_publish_tag,
    })

how can I fix it? thanks.