"""AWS Glue Catalog Module.""" from __future__ import annotations import logging from typing import TYPE_CHECKING, Any, Literal import boto3 from awswrangler import _utils, exceptions, typing from awswrangler._config import apply_configs from awswrangler.catalog._definitions import ( _csv_table_definition, _json_table_definition, _orc_table_definition, _parquet_table_definition, ) from awswrangler.catalog._delete import delete_all_partitions, delete_table_if_exists from awswrangler.catalog._get import _get_table_input from awswrangler.catalog._utils import _catalog_id, sanitize_column_name, sanitize_table_name if TYPE_CHECKING: from mypy_boto3_glue import GlueClient _logger: logging.Logger = logging.getLogger(__name__) def _update_if_necessary( dic: dict[str, str | dict[str, str]], key: str, value: str | dict[str, str] | None, mode: str ) -> str: if value is not None: if key not in dic or dic[key] != value: dic[key] = value if mode in ("append", "overwrite_partitions"): return "update" return mode def _create_table( # noqa: PLR0912,PLR0915 database: str, table: str, description: str | None, parameters: dict[str, str] | None, mode: str, catalog_versioning: bool, boto3_session: boto3.Session | None, table_input: dict[str, Any], table_exist: bool, partitions_types: dict[str, str] | None, columns_comments: dict[str, str] | None, columns_parameters: dict[str, dict[str, str]] | None, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None, catalog_id: str | None, ) -> None: # Description mode = _update_if_necessary(dic=table_input, key="Description", value=description, mode=mode) if "Parameters" not in table_input: table_input["Parameters"] = {} # Parameters parameters = parameters if parameters else {} for k, v in parameters.items(): mode = _update_if_necessary(dic=table_input["Parameters"], key=k, value=v, mode=mode) # Projection projection_params = athena_partition_projection_settings if athena_partition_projection_settings else {} if athena_partition_projection_settings: table_input["Parameters"]["projection.enabled"] = "true" partitions_types = partitions_types if partitions_types else {} projection_types = projection_params.get("projection_types", {}) projection_ranges = projection_params.get("projection_ranges", {}) projection_values = projection_params.get("projection_values", {}) projection_intervals = projection_params.get("projection_intervals", {}) projection_digits = projection_params.get("projection_digits", {}) projection_formats = projection_params.get("projection_formats", {}) projection_types = {sanitize_column_name(k): v for k, v in projection_types.items()} projection_ranges = {sanitize_column_name(k): v for k, v in projection_ranges.items()} projection_values = {sanitize_column_name(k): v for k, v in projection_values.items()} projection_intervals = {sanitize_column_name(k): v for k, v in projection_intervals.items()} projection_digits = {sanitize_column_name(k): v for k, v in projection_digits.items()} projection_formats = {sanitize_column_name(k): v for k, v in projection_formats.items()} projection_storage_location_template = projection_params.get("projection_storage_location_template") for k, v in projection_types.items(): dtype: str | None = partitions_types.get(k) if dtype is None and projection_storage_location_template is None: raise exceptions.InvalidArgumentCombination( f"Column {k} appears as projected column but not as partitioned column." ) if dtype == "date": table_input["Parameters"][f"projection.{k}.format"] = "yyyy-MM-dd" elif dtype == "timestamp": table_input["Parameters"][f"projection.{k}.format"] = "yyyy-MM-dd HH:mm:ss" table_input["Parameters"][f"projection.{k}.interval.unit"] = "SECONDS" table_input["Parameters"][f"projection.{k}.interval"] = "1" for k, v in projection_types.items(): mode = _update_if_necessary(dic=table_input["Parameters"], key=f"projection.{k}.type", value=v, mode=mode) for k, v in projection_ranges.items(): mode = _update_if_necessary(dic=table_input["Parameters"], key=f"projection.{k}.range", value=v, mode=mode) for k, v in projection_values.items(): mode = _update_if_necessary(dic=table_input["Parameters"], key=f"projection.{k}.values", value=v, mode=mode) for k, v in projection_intervals.items(): mode = _update_if_necessary( dic=table_input["Parameters"], key=f"projection.{k}.interval", value=str(v), mode=mode ) for k, v in projection_digits.items(): mode = _update_if_necessary( dic=table_input["Parameters"], key=f"projection.{k}.digits", value=str(v), mode=mode ) for k, v in projection_formats.items(): mode = _update_if_necessary( dic=table_input["Parameters"], key=f"projection.{k}.format", value=str(v), mode=mode ) mode = _update_if_necessary( table_input["Parameters"], key="storage.location.template", value=projection_storage_location_template, mode=mode, ) else: table_input["Parameters"]["projection.enabled"] = "false" # Column comments columns_comments = columns_comments if columns_comments else {} columns_comments = {sanitize_column_name(k): v for k, v in columns_comments.items()} if columns_comments: for col in table_input["StorageDescriptor"]["Columns"]: name: str = col["Name"] if name in columns_comments: mode = _update_if_necessary(dic=col, key="Comment", value=columns_comments[name], mode=mode) for par in table_input["PartitionKeys"]: name = par["Name"] if name in columns_comments: mode = _update_if_necessary(dic=par, key="Comment", value=columns_comments[name], mode=mode) # Column parameters columns_parameters = columns_parameters if columns_parameters else {} columns_parameters = {sanitize_column_name(k): v for k, v in columns_parameters.items()} if columns_parameters: for col in table_input["StorageDescriptor"]["Columns"]: name: str = col["Name"] # type: ignore[no-redef] if name in columns_parameters: mode = _update_if_necessary(dic=col, key="Parameters", value=columns_parameters[name], mode=mode) for par in table_input["PartitionKeys"]: name = par["Name"] if name in columns_parameters: mode = _update_if_necessary(dic=par, key="Parameters", value=columns_parameters[name], mode=mode) _logger.debug("table_input: %s", table_input) client_glue = _utils.client(service_name="glue", session=boto3_session) skip_archive: bool = not catalog_versioning if mode not in ("overwrite", "append", "overwrite_partitions", "update"): raise exceptions.InvalidArgument( f"{mode} is not a valid mode. It must be 'overwrite', 'append' or 'overwrite_partitions'." ) args: dict[str, Any] = _catalog_id( catalog_id=catalog_id, DatabaseName=database, TableInput=table_input, ) if table_exist: _logger.debug("Updating table (%s)...", mode) args["SkipArchive"] = skip_archive if mode == "overwrite": delete_all_partitions(table=table, database=database, catalog_id=catalog_id, boto3_session=boto3_session) if mode in ["overwrite", "update"]: client_glue.update_table(**args) else: try: _logger.debug("Creating table (%s)...", mode) client_glue.create_table(**args) except client_glue.exceptions.AlreadyExistsException: if mode == "overwrite": _utils.try_it( f=_overwrite_table, ex=client_glue.exceptions.AlreadyExistsException, client_glue=client_glue, catalog_id=catalog_id, database=database, table=table, table_input=table_input, boto3_session=boto3_session, ) _logger.debug("Leaving table as is (%s)...", mode) def _overwrite_table( client_glue: "GlueClient", catalog_id: str | None, database: str, table: str, table_input: dict[str, Any], boto3_session: boto3.Session, ) -> None: delete_table_if_exists( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id, ) args: dict[str, Any] = _catalog_id( catalog_id=catalog_id, DatabaseName=database, TableInput=table_input, ) client_glue.create_table(**args) def _upsert_table_parameters( parameters: dict[str, str], database: str, catalog_versioning: bool, catalog_id: str | None, table_input: dict[str, Any], boto3_session: boto3.Session | None, ) -> dict[str, str]: pars: dict[str, str] = table_input["Parameters"] update: bool = False for k, v in parameters.items(): if k not in pars or v != pars[k]: pars[k] = v update = True if update is True: _overwrite_table_parameters( parameters=pars, database=database, catalog_id=catalog_id, boto3_session=boto3_session, table_input=table_input, catalog_versioning=catalog_versioning, ) return pars def _overwrite_table_parameters( parameters: dict[str, str], database: str, catalog_versioning: bool, catalog_id: str | None, table_input: dict[str, Any], boto3_session: boto3.Session | None, ) -> dict[str, str]: table_input["Parameters"] = parameters client_glue = _utils.client(service_name="glue", session=boto3_session) skip_archive: bool = not catalog_versioning client_glue.update_table( **_catalog_id( catalog_id=catalog_id, DatabaseName=database, TableInput=table_input, SkipArchive=skip_archive, ) ) return parameters def _update_table_input(table_input: dict[str, Any], columns_types: dict[str, str], allow_reorder: bool = True) -> bool: column_updated = False catalog_cols: dict[str, str] = {x["Name"]: x["Type"] for x in table_input["StorageDescriptor"]["Columns"]} if not allow_reorder: for catalog_key, frame_key in zip(catalog_cols, columns_types): if catalog_key != frame_key: raise exceptions.InvalidArgumentValue(f"Column {frame_key} is out of order.") for c, t in columns_types.items(): if c not in catalog_cols: _logger.debug("New column %s with type %s.", c, t) table_input["StorageDescriptor"]["Columns"].append({"Name": c, "Type": t}) column_updated = True elif t != catalog_cols[c]: # Data type change detected! raise exceptions.InvalidArgumentValue( f"Data type change detected on column {c} (Old type: {catalog_cols[c]} / New type {t})." ) return column_updated def _create_parquet_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None, partitions_types: dict[str, str] | None, bucketing_info: typing.BucketingInfoTuple | None, catalog_id: str | None, compression: str | None, description: str | None, parameters: dict[str, str] | None, columns_comments: dict[str, str] | None, columns_parameters: dict[str, dict[str, str]] | None, mode: str, catalog_versioning: bool, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None, boto3_session: boto3.Session | None, catalog_table_input: dict[str, Any] | None, ) -> None: table = sanitize_table_name(table=table) columns_types = {sanitize_column_name(k): v for k, v in columns_types.items()} partitions_types = {} if partitions_types is None else partitions_types _logger.debug("catalog_table_input: %s", catalog_table_input) table_input: dict[str, Any] if (catalog_table_input is not None) and (mode in ("append", "overwrite_partitions")): table_input = catalog_table_input is_table_updated = _update_table_input(table_input, columns_types) if is_table_updated: mode = "update" else: table_input = _parquet_table_definition( table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, compression=compression, ) table_exist: bool = catalog_table_input is not None _logger.debug("table_exist: %s", table_exist) _create_table( database=database, table=table, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, boto3_session=boto3_session, table_input=table_input, table_exist=table_exist, partitions_types=partitions_types, athena_partition_projection_settings=athena_partition_projection_settings, catalog_id=catalog_id, ) def _create_orc_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None, partitions_types: dict[str, str] | None, bucketing_info: typing.BucketingInfoTuple | None, catalog_id: str | None, compression: str | None, description: str | None, parameters: dict[str, str] | None, columns_comments: dict[str, str] | None, columns_parameters: dict[str, dict[str, str]] | None, mode: str, catalog_versioning: bool, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None, boto3_session: boto3.Session | None, catalog_table_input: dict[str, Any] | None, ) -> None: table = sanitize_table_name(table=table) columns_types = {sanitize_column_name(k): v for k, v in columns_types.items()} partitions_types = {} if partitions_types is None else partitions_types _logger.debug("catalog_table_input: %s", catalog_table_input) table_input: dict[str, Any] if (catalog_table_input is not None) and (mode in ("append", "overwrite_partitions")): table_input = catalog_table_input is_table_updated = _update_table_input(table_input, columns_types) if is_table_updated: mode = "update" else: table_input = _orc_table_definition( table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, compression=compression, ) table_exist: bool = catalog_table_input is not None _logger.debug("table_exist: %s", table_exist) _create_table( database=database, table=table, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, boto3_session=boto3_session, table_input=table_input, table_exist=table_exist, partitions_types=partitions_types, athena_partition_projection_settings=athena_partition_projection_settings, catalog_id=catalog_id, ) def _create_csv_table( database: str, table: str, path: str | None, columns_types: dict[str, str], table_type: str | None, partitions_types: dict[str, str] | None, bucketing_info: typing.BucketingInfoTuple | None, description: str | None, compression: str | None, parameters: dict[str, str] | None, columns_comments: dict[str, str] | None, columns_parameters: dict[str, dict[str, str]] | None, mode: str, catalog_versioning: bool, schema_evolution: bool, sep: str, skip_header_line_count: int | None, serde_library: str | None, serde_parameters: dict[str, str] | None, boto3_session: boto3.Session | None, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None, catalog_table_input: dict[str, Any] | None, catalog_id: str | None, ) -> None: table = sanitize_table_name(table=table) columns_types = {sanitize_column_name(k): v for k, v in columns_types.items()} partitions_types = {} if partitions_types is None else partitions_types _logger.debug("catalog_table_input: %s", catalog_table_input) if schema_evolution is False: _utils.check_schema_changes(columns_types=columns_types, table_input=catalog_table_input, mode=mode) table_input: dict[str, Any] if (catalog_table_input is not None) and (mode in ("append", "overwrite_partitions")): table_input = catalog_table_input is_table_updated = _update_table_input(table_input, columns_types, allow_reorder=False) if is_table_updated: mode = "update" else: table_input = _csv_table_definition( table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, compression=compression, sep=sep, skip_header_line_count=skip_header_line_count, serde_library=serde_library, serde_parameters=serde_parameters, ) table_exist: bool = catalog_table_input is not None _logger.debug("table_exist: %s", table_exist) _create_table( database=database, table=table, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, boto3_session=boto3_session, table_input=table_input, table_exist=table_exist, partitions_types=partitions_types, athena_partition_projection_settings=athena_partition_projection_settings, catalog_id=catalog_id, ) def _create_json_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None, partitions_types: dict[str, str] | None, bucketing_info: typing.BucketingInfoTuple | None, description: str | None, compression: str | None, parameters: dict[str, str] | None, columns_comments: dict[str, str] | None, columns_parameters: dict[str, dict[str, str]] | None, mode: str, catalog_versioning: bool, schema_evolution: bool, serde_library: str | None, serde_parameters: dict[str, str] | None, boto3_session: boto3.Session | None, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None, catalog_table_input: dict[str, Any] | None, catalog_id: str | None, ) -> None: table = sanitize_table_name(table=table) columns_types = {sanitize_column_name(k): v for k, v in columns_types.items()} partitions_types = {} if partitions_types is None else partitions_types _logger.debug("catalog_table_input: %s", catalog_table_input) table_input: dict[str, Any] if schema_evolution is False: _utils.check_schema_changes(columns_types=columns_types, table_input=catalog_table_input, mode=mode) if (catalog_table_input is not None) and (mode in ("append", "overwrite_partitions")): table_input = catalog_table_input is_table_updated = _update_table_input(table_input, columns_types) if is_table_updated: mode = "update" else: table_input = _json_table_definition( table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, compression=compression, serde_library=serde_library, serde_parameters=serde_parameters, ) table_exist: bool = catalog_table_input is not None _logger.debug("table_exist: %s", table_exist) _create_table( database=database, table=table, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, boto3_session=boto3_session, table_input=table_input, table_exist=table_exist, partitions_types=partitions_types, athena_partition_projection_settings=athena_partition_projection_settings, catalog_id=catalog_id, ) @apply_configs def upsert_table_parameters( parameters: dict[str, str], database: str, table: str, catalog_versioning: bool = False, catalog_id: str | None = None, boto3_session: boto3.Session | None = None, ) -> dict[str, str]: """Insert or Update the received parameters. Parameters ---------- parameters e.g. {"source": "mysql", "destination": "datalake"} database Database name. table Table name. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. catalog_id The ID of the Data Catalog from which to retrieve Databases. If none is provided, the AWS account ID is used by default. boto3_session Boto3 Session. The default boto3 session will be used if boto3_session receive None. Returns ------- All parameters after the upsert. Examples -------- >>> import awswrangler as wr >>> pars = wr.catalog.upsert_table_parameters( ... parameters={"source": "mysql", "destination": "datalake"}, ... database="...", ... table="...", ... ) """ table_input: dict[str, str] | None = _get_table_input( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id, ) if table_input is None: raise exceptions.InvalidArgumentValue(f"Table {database}.{table} does not exist.") return _upsert_table_parameters( parameters=parameters, database=database, boto3_session=boto3_session, catalog_id=catalog_id, table_input=table_input, catalog_versioning=catalog_versioning, ) @apply_configs def overwrite_table_parameters( parameters: dict[str, str], database: str, table: str, catalog_versioning: bool = False, catalog_id: str | None = None, boto3_session: boto3.Session | None = None, ) -> dict[str, str]: """Overwrite all existing parameters. Parameters ---------- parameters e.g. {"source": "mysql", "destination": "datalake"} database Database name. table Table name. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. catalog_id The ID of the Data Catalog from which to retrieve Databases. If none is provided, the AWS account ID is used by default. boto3_session The default boto3 session will be used if **boto3_session** is ``None``. Returns ------- All parameters after the overwrite (The same received). Examples -------- >>> import awswrangler as wr >>> pars = wr.catalog.overwrite_table_parameters( ... parameters={"source": "mysql", "destination": "datalake"}, ... database="...", ... table="...", ... ) """ table_input: dict[str, Any] | None = _get_table_input( database=database, table=table, catalog_id=catalog_id, boto3_session=boto3_session, ) if table_input is None: raise exceptions.InvalidTable(f"Table {table} does not exist on database {database}.") return _overwrite_table_parameters( parameters=parameters, database=database, catalog_id=catalog_id, table_input=table_input, boto3_session=boto3_session, catalog_versioning=catalog_versioning, ) @apply_configs def create_database( name: str, description: str | None = None, catalog_id: str | None = None, exist_ok: bool = False, database_input_args: dict[str, Any] | None = None, boto3_session: boto3.Session | None = None, ) -> None: """Create a database in AWS Glue Catalog. Parameters ---------- name Database name. description A description for the Database. catalog_id The ID of the Data Catalog from which to retrieve Databases. If ``None`` is provided, the AWS account ID is used by default. exist_ok If set to ``True`` will not raise an Exception if a Database with the same already exists. In this case the description will be updated if it is different from the current one. database_input_args Additional metadata to pass to database creation. Supported arguments listed here: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/glue.html#Glue.Client.create_database boto3_session The default boto3 session will be used if **boto3_session** receive ``None``. Examples -------- >>> import awswrangler as wr >>> wr.catalog.create_database( ... name='awswrangler_test' ... ) """ client_glue = _utils.client(service_name="glue", session=boto3_session) args: dict[str, Any] = {"Name": name, **database_input_args} if database_input_args else {"Name": name} if description is not None: args["Description"] = description try: r = client_glue.get_database(**_catalog_id(catalog_id=catalog_id, Name=name)) if not exist_ok: raise exceptions.AlreadyExists(f"Database {name} already exists and is set to False.") for k, v in args.items(): if v != r["Database"].get(k, ""): client_glue.update_database(**_catalog_id(catalog_id=catalog_id, Name=name, DatabaseInput=args)) break except client_glue.exceptions.EntityNotFoundException: client_glue.create_database(**_catalog_id(catalog_id=catalog_id, DatabaseInput=args)) @apply_configs def create_parquet_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None = None, partitions_types: dict[str, str] | None = None, bucketing_info: typing.BucketingInfoTuple | None = None, catalog_id: str | None = None, compression: str | None = None, description: str | None = None, parameters: dict[str, str] | None = None, columns_comments: dict[str, str] | None = None, columns_parameters: dict[str, dict[str, str]] | None = None, mode: Literal["overwrite", "append"] = "overwrite", catalog_versioning: bool = False, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None = None, boto3_session: boto3.Session | None = None, ) -> None: """Create a Parquet Table (Metadata Only) in the AWS Glue Catalog. 'https://docs.aws.amazon.com/athena/latest/ug/data-types.html' Parameters ---------- database Database name. table Table name. path Amazon S3 path (e.g. s3://bucket/prefix/). columns_types Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}). table_type The type of the Glue Table. Set to ``EXTERNAL_TABLE`` if ``None``. partitions_types Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}). bucketing_info Tuple consisting of the column names used for bucketing as the first element and the number of buckets as the second element. Only `str`, `int` and `bool` are supported as column data types for bucketing. catalog_id The ID of the Data Catalog from which to retrieve Databases. If none is provided, the AWS account ID is used by default. compression Compression style (``None``, ``snappy``, ``gzip``, etc). description Table description parameters Key/value pairs to tag the table. columns_comments Columns names and the related comments (e.g. {'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'}). columns_parameters Columns names and the related parameters (e.g. {'col0': {'par0': 'Param 0', 'par1': 'Param 1'}}). mode 'overwrite' to recreate any possible existing table or 'append' to keep any possible existing table. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. athena_partition_projection_settings Parameters of the Athena Partition Projection (https://docs.aws.amazon.com/athena/latest/ug/partition-projection.html). AthenaPartitionProjectionSettings is a `TypedDict`, meaning the passed parameter can be instantiated either as an instance of AthenaPartitionProjectionSettings or as a regular Python dict. Following projection parameters are supported: .. list-table:: Projection Parameters :header-rows: 1 * - Name - Type - Description * - projection_types - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections types. Valid types: "enum", "integer", "date", "injected" https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'enum', 'col2_name': 'integer'}) * - projection_ranges - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections ranges. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '0,10', 'col2_name': '-1,8675309'}) * - projection_values - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections values. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'A,B,Unknown', 'col2_name': 'foo,boo,bar'}) * - projection_intervals - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections intervals. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '5'}) * - projection_digits - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections digits. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '2'}) * - projection_formats - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections formats. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_date': 'yyyy-MM-dd', 'col2_timestamp': 'yyyy-MM-dd HH:mm:ss'}) * - projection_storage_location_template - Optional[str] - Value which is allows Athena to properly map partition values if the S3 file locations do not follow a typical `.../column=value/...` pattern. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-setting-up.html (e.g. s3://bucket/table_root/a=${a}/${b}/some_static_subdirectory/${c}/) boto3_session The default boto3 session will be used if **boto3_session** receive ``None``. Examples -------- >>> import awswrangler as wr >>> wr.catalog.create_parquet_table( ... database='default', ... table='my_table', ... path='s3://bucket/prefix/', ... columns_types={'col0': 'bigint', 'col1': 'double'}, ... partitions_types={'col2': 'date'}, ... compression='snappy', ... description='My own table!', ... parameters={'source': 'postgresql'}, ... columns_comments={'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'} ... ) """ catalog_table_input: dict[str, Any] | None = _get_table_input( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id, ) _create_parquet_table( database=database, table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, catalog_id=catalog_id, compression=compression, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, athena_partition_projection_settings=athena_partition_projection_settings, boto3_session=boto3_session, catalog_table_input=catalog_table_input, ) @apply_configs def create_orc_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None = None, partitions_types: dict[str, str] | None = None, bucketing_info: typing.BucketingInfoTuple | None = None, catalog_id: str | None = None, compression: str | None = None, description: str | None = None, parameters: dict[str, str] | None = None, columns_comments: dict[str, str] | None = None, columns_parameters: dict[str, dict[str, str]] | None = None, mode: Literal["overwrite", "append"] = "overwrite", catalog_versioning: bool = False, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None = None, boto3_session: boto3.Session | None = None, ) -> None: """Create a ORC Table (Metadata Only) in the AWS Glue Catalog. 'https://docs.aws.amazon.com/athena/latest/ug/data-types.html' Parameters ---------- database Database name. table Table name. path Amazon S3 path (e.g. s3://bucket/prefix/). columns_types Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}). table_type The type of the Glue Table. Set to EXTERNAL_TABLE if None. partitions_types Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}). bucketing_info Tuple consisting of the column names used for bucketing as the first element and the number of buckets as the second element. Only `str`, `int` and `bool` are supported as column data types for bucketing. catalog_id The ID of the Data Catalog from which to retrieve Databases. If none is provided, the AWS account ID is used by default. compression Compression style (``None``, ``snappy``, ``gzip``, etc). description Table description parameters Key/value pairs to tag the table. columns_comments Columns names and the related comments (e.g. {'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'}). columns_parameters Columns names and the related parameters (e.g. {'col0': {'par0': 'Param 0', 'par1': 'Param 1'}}). mode 'overwrite' to recreate any possible existing table or 'append' to keep any possible existing table. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. athena_partition_projection_settings Parameters of the Athena Partition Projection (https://docs.aws.amazon.com/athena/latest/ug/partition-projection.html). AthenaPartitionProjectionSettings is a `TypedDict`, meaning the passed parameter can be instantiated either as an instance of AthenaPartitionProjectionSettings or as a regular Python dict. Following projection parameters are supported: .. list-table:: Projection Parameters :header-rows: 1 * - Name - Type - Description * - projection_types - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections types. Valid types: "enum", "integer", "date", "injected" https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'enum', 'col2_name': 'integer'}) * - projection_ranges - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections ranges. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '0,10', 'col2_name': '-1,8675309'}) * - projection_values - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections values. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'A,B,Unknown', 'col2_name': 'foo,boo,bar'}) * - projection_intervals - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections intervals. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '5'}) * - projection_digits - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections digits. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '2'}) * - projection_formats - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections formats. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_date': 'yyyy-MM-dd', 'col2_timestamp': 'yyyy-MM-dd HH:mm:ss'}) * - projection_storage_location_template - Optional[str] - Value which is allows Athena to properly map partition values if the S3 file locations do not follow a typical `.../column=value/...` pattern. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-setting-up.html (e.g. s3://bucket/table_root/a=${a}/${b}/some_static_subdirectory/${c}/) boto3_session The default boto3 session will be used if **boto3_session** receive ``None``. Examples -------- >>> import awswrangler as wr >>> wr.catalog.create_parquet_table( ... database='default', ... table='my_table', ... path='s3://bucket/prefix/', ... columns_types={'col0': 'bigint', 'col1': 'double'}, ... partitions_types={'col2': 'date'}, ... compression='snappy', ... description='My own table!', ... parameters={'source': 'postgresql'}, ... columns_comments={'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'} ... ) """ catalog_table_input: dict[str, Any] | None = _get_table_input( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id, ) _create_orc_table( database=database, table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, catalog_id=catalog_id, compression=compression, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, athena_partition_projection_settings=athena_partition_projection_settings, boto3_session=boto3_session, catalog_table_input=catalog_table_input, ) @apply_configs def create_csv_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None = None, partitions_types: dict[str, str] | None = None, bucketing_info: typing.BucketingInfoTuple | None = None, compression: str | None = None, description: str | None = None, parameters: dict[str, str] | None = None, columns_comments: dict[str, str] | None = None, columns_parameters: dict[str, dict[str, str]] | None = None, mode: Literal["overwrite", "append"] = "overwrite", catalog_versioning: bool = False, schema_evolution: bool = False, sep: str = ",", skip_header_line_count: int | None = None, serde_library: str | None = None, serde_parameters: dict[str, str] | None = None, boto3_session: boto3.Session | None = None, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None = None, catalog_id: str | None = None, ) -> None: r"""Create a CSV Table (Metadata Only) in the AWS Glue Catalog. 'https://docs.aws.amazon.com/athena/latest/ug/data-types.html' Note ---- Athena requires the columns in the underlying CSV files in S3 to be in the same order as the columns in the Glue data catalog. Parameters ---------- database Database name. table Table name. path Amazon S3 path (e.g. s3://bucket/prefix/). columns_types Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}). table_type The type of the Glue Table. Set to EXTERNAL_TABLE if None. partitions_types Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}). bucketing_info Tuple consisting of the column names used for bucketing as the first element and the number of buckets as the second element. Only `str`, `int` and `bool` are supported as column data types for bucketing. compression Compression style (``None``, ``gzip``, etc). description Table description parameters Key/value pairs to tag the table. columns_comments Columns names and the related comments (e.g. {'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'}). columns_parameters Columns names and the related parameters (e.g. {'col0': {'par0': 'Param 0', 'par1': 'Param 1'}}). mode 'overwrite' to recreate any possible existing table or 'append' to keep any possible existing table. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. schema_evolution If True allows schema evolution (new or missing columns), otherwise a exception will be raised. (Only considered if dataset=True and mode in ("append", "overwrite_partitions")) Related tutorial: https://aws-sdk-pandas.readthedocs.io/en/3.11.0/tutorials/014%20-%20Schema%20Evolution.html sep String of length 1. Field delimiter for the output file. skip_header_line_count Number of Lines to skip regarding to the header. serde_library Specifies the SerDe Serialization library which will be used. You need to provide the Class library name as a string. If no library is provided the default is `org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe`. serde_parameters Dictionary of initialization parameters for the SerDe. The default is `{"field.delim": sep, "escape.delim": "\\"}`. athena_partition_projection_settings Parameters of the Athena Partition Projection (https://docs.aws.amazon.com/athena/latest/ug/partition-projection.html). AthenaPartitionProjectionSettings is a `TypedDict`, meaning the passed parameter can be instantiated either as an instance of AthenaPartitionProjectionSettings or as a regular Python dict. Following projection parameters are supported: .. list-table:: Projection Parameters :header-rows: 1 * - Name - Type - Description * - projection_types - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections types. Valid types: "enum", "integer", "date", "injected" https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'enum', 'col2_name': 'integer'}) * - projection_ranges - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections ranges. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '0,10', 'col2_name': '-1,8675309'}) * - projection_values - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections values. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'A,B,Unknown', 'col2_name': 'foo,boo,bar'}) * - projection_intervals - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections intervals. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '5'}) * - projection_digits - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections digits. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '2'}) * - projection_formats - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections formats. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_date': 'yyyy-MM-dd', 'col2_timestamp': 'yyyy-MM-dd HH:mm:ss'}) * - projection_storage_location_template - Optional[str] - Value which is allows Athena to properly map partition values if the S3 file locations do not follow a typical `.../column=value/...` pattern. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-setting-up.html (e.g. s3://bucket/table_root/a=${a}/${b}/some_static_subdirectory/${c}/) boto3_session The default boto3 session will be used if **boto3_session** receive ``None``. catalog_id The ID of the Data Catalog from which to retrieve Databases. If ``None`` is provided, the AWS account ID is used by default. Examples -------- >>> import awswrangler as wr >>> wr.catalog.create_csv_table( ... database='default', ... table='my_table', ... path='s3://bucket/prefix/', ... columns_types={'col0': 'bigint', 'col1': 'double'}, ... partitions_types={'col2': 'date'}, ... compression='gzip', ... description='My own table!', ... parameters={'source': 'postgresql'}, ... columns_comments={'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'} ... ) """ catalog_table_input: dict[str, Any] | None = _get_table_input( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id, ) _create_csv_table( database=database, table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, catalog_id=catalog_id, compression=compression, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, schema_evolution=schema_evolution, athena_partition_projection_settings=athena_partition_projection_settings, boto3_session=boto3_session, catalog_table_input=catalog_table_input, sep=sep, skip_header_line_count=skip_header_line_count, serde_library=serde_library, serde_parameters=serde_parameters, ) @apply_configs def create_json_table( database: str, table: str, path: str, columns_types: dict[str, str], table_type: str | None = None, partitions_types: dict[str, str] | None = None, bucketing_info: typing.BucketingInfoTuple | None = None, compression: str | None = None, description: str | None = None, parameters: dict[str, str] | None = None, columns_comments: dict[str, str] | None = None, columns_parameters: dict[str, dict[str, str]] | None = None, mode: Literal["overwrite", "append"] = "overwrite", catalog_versioning: bool = False, schema_evolution: bool = False, serde_library: str | None = None, serde_parameters: dict[str, str] | None = None, boto3_session: boto3.Session | None = None, athena_partition_projection_settings: typing.AthenaPartitionProjectionSettings | None = None, catalog_id: str | None = None, ) -> None: r"""Create a JSON Table (Metadata Only) in the AWS Glue Catalog. 'https://docs.aws.amazon.com/athena/latest/ug/data-types.html' Parameters ---------- database Database name. table Table name. path Amazon S3 path (e.g. s3://bucket/prefix/). columns_types Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}). table_type The type of the Glue Table. Set to EXTERNAL_TABLE if None. partitions_types Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}). bucketing_info Tuple consisting of the column names used for bucketing as the first element and the number of buckets as the second element. Only `str`, `int` and `bool` are supported as column data types for bucketing. compression Compression style (``None``, ``gzip``, etc). description Table description parameters Key/value pairs to tag the table. columns_comments Columns names and the related comments (e.g. {'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'}). columns_parameters Columns names and the related parameters (e.g. {'col0': {'par0': 'Param 0', 'par1': 'Param 1'}}). mode 'overwrite' to recreate any possible existing table or 'append' to keep any possible existing table. catalog_versioning If True and `mode="overwrite"`, creates an archived version of the table catalog before updating it. schema_evolution If True allows schema evolution (new or missing columns), otherwise a exception will be raised. (Only considered if dataset=True and mode in ("append", "overwrite_partitions")) Related tutorial: https://aws-sdk-pandas.readthedocs.io/en/3.11.0/tutorials/014%20-%20Schema%20Evolution.html serde_library Specifies the SerDe Serialization library which will be used. You need to provide the Class library name as a string. If no library is provided the default is `org.openx.data.jsonserde.JsonSerDe`. serde_parameters Dictionary of initialization parameters for the SerDe. The default is `{"field.delim": sep, "escape.delim": "\\"}`. athena_partition_projection_settings Parameters of the Athena Partition Projection (https://docs.aws.amazon.com/athena/latest/ug/partition-projection.html). AthenaPartitionProjectionSettings is a `TypedDict`, meaning the passed parameter can be instantiated either as an instance of AthenaPartitionProjectionSettings or as a regular Python dict. Following projection parameters are supported: .. list-table:: Projection Parameters :header-rows: 1 * - Name - Type - Description * - projection_types - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections types. Valid types: "enum", "integer", "date", "injected" https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'enum', 'col2_name': 'integer'}) * - projection_ranges - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections ranges. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '0,10', 'col2_name': '-1,8675309'}) * - projection_values - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections values. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': 'A,B,Unknown', 'col2_name': 'foo,boo,bar'}) * - projection_intervals - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections intervals. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '5'}) * - projection_digits - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections digits. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_name': '1', 'col2_name': '2'}) * - projection_formats - Optional[Dict[str, str]] - Dictionary of partitions names and Athena projections formats. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-supported-types.html (e.g. {'col_date': 'yyyy-MM-dd', 'col2_timestamp': 'yyyy-MM-dd HH:mm:ss'}) * - projection_storage_location_template - Optional[str] - Value which is allows Athena to properly map partition values if the S3 file locations do not follow a typical `.../column=value/...` pattern. https://docs.aws.amazon.com/athena/latest/ug/partition-projection-setting-up.html (e.g. s3://bucket/table_root/a=${a}/${b}/some_static_subdirectory/${c}/) boto3_session The default boto3 session will be used if **boto3_session** receive ``None``. catalog_id The ID of the Data Catalog from which to retrieve Databases. If ``None`` is provided, the AWS account ID is used by default. Examples -------- >>> import awswrangler as wr >>> wr.catalog.create_json_table( ... database='default', ... table='my_table', ... path='s3://bucket/prefix/', ... columns_types={'col0': 'bigint', 'col1': 'double'}, ... partitions_types={'col2': 'date'}, ... description='My very own JSON table!', ... parameters={'source': 'postgresql'}, ... columns_comments={'col0': 'Column 0.', 'col1': 'Column 1.', 'col2': 'Partition.'} ... ) """ catalog_table_input: dict[str, Any] | None = _get_table_input( database=database, table=table, boto3_session=boto3_session, catalog_id=catalog_id ) _create_json_table( database=database, table=table, path=path, columns_types=columns_types, table_type=table_type, partitions_types=partitions_types, bucketing_info=bucketing_info, catalog_id=catalog_id, compression=compression, description=description, parameters=parameters, columns_comments=columns_comments, columns_parameters=columns_parameters, mode=mode, catalog_versioning=catalog_versioning, schema_evolution=schema_evolution, athena_partition_projection_settings=athena_partition_projection_settings, boto3_session=boto3_session, catalog_table_input=catalog_table_input, serde_library=serde_library, serde_parameters=serde_parameters, )