Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Your message was too long by 963 tokens - please shorten it and try again.

 For reference, here's your message at the allowed limit:
&gt; "@task(name="process_chunk")
def process_chunk(chunk, table_name, athena_schema_dict, raw_bucket, db_name, today_date):
    logger = get_run_logger()
    chunk['id_jour'] = today_date
    <http://logger.info|logger.info>(f"Processing chunk for {table_name}")

    for col in chunk.columns:
        <http://logger.info|logger.info>(f"Processing column '{col}' of type {chunk[col].dtype}")

        if col == 'id_jour':
            <http://logger.info|logger.info>(f"Skipping conversion for column '{col}'")
            continue

        athena_type = athena_schema_dict.get(col.lower())
        <http://logger.info|logger.info>(f"Athena type for column '{col}': {athena_type}")

        # Convert based on Athena type if necessary
        if athena_type == 'timestamp':
            try:
                chunk[col] = <http://pd.to|pd.to>_datetime(chunk[col], errors='coerce')
                <http://logger.info|logger.info>(f"Column {col} successfully converted to datetime.")
            except Exception as e:
                logger.error(f"Error converting column {col} to datetime: {e}")

        elif athena_type == 'bigint':
            try:
                chunk[col] = <http://pd.to|pd.to>_numeric(chunk[col], errors='coerce', downcast='