IBM watsonx.data

# Improves performance by setting the target table to regularly remove 
# old metadata files. 
#
# First, install the following dependencies into your Python virtual 
# environment:
# 
# pip install requests pyiceberg pyarrow
#
# Then, set the following environment variables:
#
# IBM_IAM_API_KEY - An API key value for the target IBM Cloud account.
# IBM_ICEBERG_CATALOG_METASTORE_REST_ENDPOINT - The metastore REST endpoint 
#     value for the target Apache Iceberg catalog in the target IBM watsonx.data 
#     data store instance.
# IBM_COS_BUCKET_PUBLIC_ENDPOINT - The target IBM Cloud Object Storage (COS) 
#     instance’s endpoint value.
# IBM_COS_ACCESS_KEY - An HMAC access key ID for the target COS instance.
# IBM_COS_SECRET_ACCESS_KEY - The associated HMAC secret access key ID for the 
#     target HMAC access key.
# IBM_COS_BUCKET_REGION - The target COS instance’s region short ID.
# IBM_ICEBERG_CATALOG - The name of the target Iceberg catalog.
# IBM_ICEBERG_SCHEMA - The name of the target namespace (also known as a schema) 
#     in the target catalog.
# IBM_ICEBERG_TABLE - The name of the target table in the target schema.
#
# To get these values, see the Unstructured documentation for the 
#     IBM watsonx.data connector.

import os
import requests
from pyiceberg.catalog import load_catalog

def main():
   # Get a bearer token for the target IBM Cloud account.   
   bearer_token = requests.post(
      url="https://iam.cloud.ibm.com/identity/token",
      headers={
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "application/json"
      },
      data={
            "grant_type": "urn:ibm:params:oauth:grant-type:apikey", 
            "apikey": os.getenv("IBM_IAM_API_KEY")
      }
   ).json().get("access_token")

   # Connect to the target Iceberg catalog.
   catalog = load_catalog(
      os.getenv("IBM_ICEBERG_CATALOG"),
      **{
            "type": "rest",
            "uri": f"https://{os.getenv("IBM_ICEBERG_CATALOG_METASTORE_REST_ENDPOINT")}/mds/iceberg",
            "token": bearer_token,
            "warehouse": os.getenv("IBM_ICEBERG_CATALOG"),
            "s3.endpoint": os.getenv("IBM_COS_BUCKET_PUBLIC_ENDPOINT"),
            "s3.access-key-id": os.getenv("IBM_COS_ACCESS_KEY"),
            "s3.secret-access-key": os.getenv("IBM_COS_SECRET_ACCESS_KEY"),
            "s3.region": os.getenv("IBM_COS_BUCKET_REGION")
      },
   )
            
   # Load the target table.
   table = catalog.load_table(f"{os.getenv("IBM_ICEBERG_SCHEMA")}.{os.getenv("IBM_ICEBERG_TABLE")}")

   # Set the target table's properties to remove old metadata files.
   with table.transaction() as transaction:
      transaction.set_properties(
            {
               "commit.manifest.min-count-to-merge": 10,
               "commit.manifest-merge.enabled": True,
               "write.metadata.previous-versions-max": 10,
               "write.metadata.delete-after-commit.enabled": True,
            }
      )

   # Confirm that the target table's properties were set as expected.
   print(table.metadata.properties)

if __name__ == "__main__":
   main()