Каталог Pyiceberg в GCS: я не могу использовать pyceberg с облачным хранилищем Google

Pyiceberg — библиотека Python для работы с таблицами Iceberg.

Сначала получите токен OAuth2, используя файл учетной записи службы. Я запускаю это в совместной работе, поэтому мне нужно было сделать это таким образом. Вы можете сделать это по-другому, если работаете в контейнере.

import google.auth
from google.auth.transport.requests import Request
from pyiceberg import catalog


def get_access_token(service_account_file, scopes):
  """
  Retrieves an access token from Google Cloud Platform using service account credentials.

  Args:
      service_account_file: Path to the service account JSON key file.
      scopes: List of OAuth scopes required for your application.

  Returns:
      The access token as a string.
  """

  credentials, name = google.auth.load_credentials_from_file(
      service_account_file, scopes=scopes)

  request = Request()
  credentials.refresh(request)  # Forces token refresh if needed
  return credentials

# Example usage
service_account_file = "/path-to-service-account-file.json"  # Replace with your path
scopes = ["https://www.googleapis.com/auth/cloud-platform"]  # Adjust scopes as needed

access_token = get_access_token(service_account_file, scopes)

Далее загружается каталог. Мы используем учетные данные OAUTH2, полученные с помощью ключа нашей сервисной учетной записи.

Я отредактировал функцию datetime_to_unix_ms, чтобы сосредоточиться на основной задаче.

Поскольку вы только начинаете, я предлагаю упростить вашу реализацию, используя базу данных для вашего реестра.

Если у вас уже есть кластер EMR, вам следует вместо этого рассмотреть возможность использования метахранилища Hive.

В этом примере мы будем использовать базу данных sqlite для нашего центрального реестра. Вы можете заменить его любым из параметров базы данных SQL, поддерживаемых библиотекой SQLalchemy.

REGISTRY_DATABASE_URI = "sqlite:///catalog.db" # replace this with your database URI

catalog_inst = catalog.load_catalog(
    "default",
    **{
        "uri": REGISTRY_DATABASE_URI, 
        "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO",
        "gcs.oauth2.token-expires-at": datetime_to_unix_ms(access_token.expiry),
        "gcs.project-id": "project-id", # replace with your gcp project id
        "gcs.oauth2.token": access_token.token,
        "gcs.default-bucket-location": "gs://bucket/", # replace with your gcs bucket
        "warehouse": "gs://bucket/" # replace with your gcs bucket
    }
)

Наконец, мы создаем пример таблицы с некоторыми данными, используя Pyarrow:

import pyarrow as pa

catalog_inst.create_namespace("default") # Replace this with your namespace

# Define the schema for the book table
schema = pa.schema([
    ('title', pa.string())
])

catalog_inst.drop_table("default.books") # Replace this with your table
table = catalog_inst.create_table("default.books", schema=schema)

# Create some sample data
titles = ["The Lord of the Rings", "Pride and Prejudice", "Moby Dick"]

# Create Arrow arrays from the data
title_array = pa.array(titles, type=pa.string())
table_data = pa.Table.from_arrays([title_array], names=schema.names)

table.append(table_data)