def METHOD_NAME(self, spark: SparkSession, args: Dict[str, Any]) -> None:
    logger: Logger = self.get_logger(spark=spark)
    ignore_keys = {constants.KAFKA_BOOTSTRAP_SERVERS}
    filtered_args = {key:val for key,val in args.items() if key not in ignore_keys}
    logger.info(
        "Starting Kafka to Bigquery Pyspark job with parameters:\n"
        f"{pprint.pformat(filtered_args)}"
    )
    #arguments
    bootstrap_server_list: str = args[constants.KAFKA_BOOTSTRAP_SERVERS]
    checkpoint_location: str = args[constants.KAFKA_BQ_CHECKPOINT_LOCATION]
    kafka_topics: str= args[constants.KAFKA_BQ_TOPIC]
    big_query_dataset: str = args[constants.KAFKA_BQ_DATASET]
    big_query_table: str = args[constants.KAFKA_BQ_TABLE_NAME]
    bq_temp_bucket: str = args[constants.KAFKA_BQ_TEMP_BUCKET_NAME]
    timeout: int = int(args[constants.KAFKA_BQ_TERMINATION_TIMEOUT])
    offset:str = args[constants.KAFKA_BQ_STARTING_OFFSET]
    output_mode = args[constants.KAFKA_BQ_OUTPUT_MODE]
    df = spark.readStream.format(constants.KAFKA_INPUT_FORMAT) \
              .option('kafka.bootstrap.servers', bootstrap_server_list) \
              .option('subscribe', kafka_topics) \
              .option('startingOffsets',offset) \
              .load()
    df = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    # Write
    output = df \
    .writeStream \
    .format(constants.FORMAT_BIGQUERY) \
    .outputMode(output_mode) \
    .option('checkpointLocation',checkpoint_location) \
    .option('table',big_query_dataset+'.'+big_query_table) \
    .option('temporaryGcsBucket', bq_temp_bucket) \
    .start() 
    output.awaitTermination(timeout)