Function upload_df_to_athena

sdk/python/feast/infra/utils/aws_utils.py:954–1001

Uploads a Pandas DataFrame to S3(Athena) as a new table. The caller is responsible for deleting the table when no longer necessary. Args: athena_client: Athena API Service client data_source: Athena Data Source database: Athena Database Name workgroup: Athen

(
    athena_client,
    data_source: str,
    database: str,
    workgroup: str,
    s3_resource,
    s3_path: str,
    table_name: str,
    df: pd.DataFrame,
)

def upload_df_to_athena(
    athena_client,
    data_source: str,
    database: str,
    workgroup: str,
    s3_resource,
    s3_path: str,
    table_name: str,
    df: pd.DataFrame,
):
    """Uploads a Pandas DataFrame to S3(Athena) as a new table.

    The caller is responsible for deleting the table when no longer necessary.

    Args:
        athena_client: Athena API Service client
        data_source: Athena Data Source
        database: Athena Database Name
        workgroup: Athena Workgroup Name
        s3_resource: S3 Resource object
        s3_path: S3 path where the Parquet file is temporarily uploaded
        table_name: The name of the new Data Catalog table where we copy the dataframe
        df: The Pandas DataFrame to upload

    Raises:
        AthenaTableNameTooLong: The specified table name is too long.
    """

    # Drop the index so that we dont have unnecessary columns
    df.reset_index(drop=True, inplace=True)

    # Convert Pandas DataFrame into PyArrow table and compile the Athena table schema.
    # Note, if the underlying data has missing values,
    # pandas will convert those values to np.nan if the dtypes are numerical (floats, ints, etc.) or boolean.
    # If the dtype is 'object', then missing values are inferred as python `None`s.
    # More details at:
    # https://pandas.pydata.org/pandas-docs/stable/user_guide/missing_data.html#values-considered-missing
    table = pa.Table.from_pandas(df)
    upload_arrow_table_to_athena(
        table,
        athena_client,
        data_source=data_source,
        database=database,
        workgroup=workgroup,
        s3_resource=s3_resource,
        s3_path=s3_path,
        table_name=table_name,
    )


def upload_arrow_table_to_athena(