ການຝຶກອົບຮົມ Data Engineer ໃນ PyIceberg

ຊື່ຫຍໍ້ຂອງ : Diptiman Raichaudhuri, Staff Developer Advocate at Confluent

ຄໍາຖາມນີ້ສະແດງໃຫ້ເຫັນວິສະວະກອນຂໍ້ມູນກ່ຽວກັບວິທີການນໍາໃຊ້ PyIceberg, ວິສະວະກອນ Python ນ້ໍາຫນັກແລະເຂັ້ມແຂງ. PyIceberg ເຮັດໃຫ້ໄດ້ຢ່າງງ່າຍດາຍທີ່ຈະເຮັດວຽກຂອງຂໍ້ມູນທົ່ວໄປເຊັ່ນ: ການສ້າງ, ການອ່ານ, ການປ່ຽນແປງ, ຫຼືການແກ້ໄຂຂໍ້ມູນໃນ Apache Iceberg, ບໍ່ຈໍາເປັນຕ້ອງມີ cluster ຂະຫນາດໃຫຍ່.

ປະຕິບັດຕາມຄວາມຕ້ອງການທຸລະກິດທີ່ເຫມາະສົມແລະຄວາມຕ້ອງການທີ່ຈະທົດສອບຂະຫນາດໃຫຍ່ຂອງຂໍ້ມູນ, platforms data ໄດ້ປ່ຽນແປງຢ່າງກວ້າງຂວາງໃນໄລຍະປີທີ່ຜ່ານມາເພື່ອຊ່ວຍໃຫ້ບໍລິສັດຊອກຫາຄວາມຮູ້ແລະຄ່າໃຊ້ຈ່າຍເພີ່ມເຕີມຈາກແຜ່ນຂໍ້ມູນຕ່າງໆຂອງເຂົາເຈົ້າ.

ສໍາລັບການນໍາໃຊ້ການທົດສອບອຸດສາຫະກໍາ, ສະຫນັບສະຫນູນ Lakehouse Open Data ແມ່ນໃນປັດຈຸບັນໃນປັດຈຸບັນໃນປັດຈຸບັນໃນປັດຈຸບັນ. ສະຫນັບສະຫນູນ Lakehouse Open Data ອະນຸຍາດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິສັດໃຫ້ບໍລິການ.

ວິທະຍາໄລຂອງ Composable Data Platform

ສະຫນັບສະຫນູນຂໍ້ມູນທີ່ສ້າງຕັ້ງຂຶ້ນດ້ວຍ Apache Iceberg ມີສາມຊຸດ:

1. ເອກະສານຂໍ້ມູນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນອຸປະກອນ
2. ປະເພດ metadata: ໃນປັດຈຸບັນ, ໃນປັດຈຸບັນ, Iceberg ໄດ້ຮັບການຄຸ້ມຄອງບັນຊີລາຍຊື່ຂອງເອກະສານ. The metadata layer manages table snapshots, schema, and file statistics.
3. Iceberg Catalog: Catalogues ເປັນສະຖາບັນພື້ນຖານທີ່ອະນຸຍາດໃຫ້ການຊອກຫາ, ການສ້າງ, ແລະການປ່ຽນແປງຕາຕະລາງແລະປະກັນຜົນປະໂຫຍດຂອງການຄຸ້ມຄອງຕາຕະລາງ Iceberg.

ຮູບພາບ ສໍາ ລັບ Apache Iceberg:

ລະຫັດ QR ທີ່ຢູ່ ສະ ຫນັບ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ (ລະຫັດ QR)

ເປັນຫຍັງ PyIceberg?

PyIceberg ສະຫນັບສະຫນູນວິສະວະກໍາແລະວິສະວະກອນຂໍ້ມູນເພື່ອສ້າງໂຄງປະກອບການຂົນສົ່ງທີ່ເຫມາະສົມໃນໄລຍະຫຼາຍຂອງ clouds (AWS, Azure, ແລະ Google Cloud) ແລະ storage on-premise. PyIceberg ແມ່ນການນໍາໃຊ້ Python ສໍາລັບການເຂົ້າເຖິງ tables Iceberg. ຜູ້ພັດທະນາທີ່ໃຊ້ PyIceberg ສາມາດນໍາໃຊ້ການປ່ຽນແປງຂໍ້ມູນ Pythonic ໂດຍບໍ່ຈໍາເປັນຕ້ອງໃຊ້ເຄື່ອງຊອບແວປະສິດທິພາບສູງໃນ Java Virtual Machine (JVM) clusters. PyIceberg ໃຊ້ catalogues ເພື່ອດາວໂຫລດ tables Iceberg ແລະເຮັດການປະຕິບັດ read-writing-upsert. ມັນປິ່ນປົວ aspects metadata ແລະ format table ຂອງ Iceberg, ເຮັດໃຫ້ວິສະວະກອນຂໍ້ມູນສາມາດນໍາໃຊ້

PyIceberg ສາມາດເຮັດວຽກເປັນໂຄງການ standalone ຫຼືໃນ clusters Kubernetes ສໍາລັບການ tolerance ຄວາມຜິດພາດ. ການເຊື່ອມຕໍ່ຂອງຕົນກັບ protocols Iceberg catalogues ເຊັ່ນ REST, SQLCatalog, ຫຼື AWS Glue, ເຮັດໃຫ້ມັນເປັນການເລືອກທີ່ຍິ່ງໃຫຍ່ແລະງ່າຍສໍາລັບການ querying Iceberg ຕາຕະລາງບໍ່ມີຄວາມຕ້ອງການຂອງ clusters JVM / py4j.

ການຕິດຕັ້ງການຜະລິດຂອງ PyIceberg ມີຜົນກະທົບຢ່າງກວ້າງຂວາງໃນການຕິດຕັ້ງການເຮັດວຽກຂອງການຂົນສົ່ງຂໍ້ມູນເຊັ່ນ:ລະຫັດ QRນີ້ແມ່ນເຄື່ອງມືທີ່ເຂັ້ມແຂງເພື່ອປັບປຸງຄວາມຮ້ອນລະຫວ່າງລະບົບຂໍ້ມູນການປະຕິບັດແລະລະບົບຂໍ້ມູນການວິເຄາະ.

ວິທີການ PyIceberg

PyIceberg ສະ ຫນັບ ສະ ຫນັບ ສະ ຫນັບ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ ສະ ຫນູນ

ໃນພາກສ່ວນຕໍ່ໄປ, ກະລຸນາສ້າງ demo ກັບ PyIceberg ເພື່ອຮູ້ວິທີການເຮັດວຽກຂອງຮູບແບບການອ່ານແລະພິມຂອງ Iceberg.

ວິທີການສ້າງ Lakehouse Data ມີ PyIceberg, PyArrow ແລະ DuckDB

ພວກເຮົາຈະນໍາໃຊ້ PyIceberg ແລະ PyArrow ເພື່ອ insert / upert ແລະ delete Iceberg data ແລະສ້າງໃນ Visual Studio Code (VS Code).

ຫນ້າທໍາອິດ, ການສະພາບແວດລ້ອມ virtual Python ໃຫມ່ ໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນທີ່ຊື່ວ່າ 'pyiceberg_playground', ໂດຍໃຊ້ການຄຸ້ມຄອງຄໍາແນະນໍານີ້:

$>python -m venv iceberg_playground

ຫຼັງຈາກນັ້ນ directory ນີ້ - 'iceberg_playground' - 'ແມ່ນເປີດໃນ VS Code ທີ່ໂຄງການ PyIceberg ຈະໄດ້ຮັບການ hosted. ຮູບພາບທີ່ຜ່ານມາສະແດງໃຫ້ເຫັນ VS Code 'clean slate'.

ຫຼັງຈາກນັ້ນ PyIceberg ແລະສະບັບອື່ນໆຖືກຕິດຕັ້ງໃນສະພາບແວດລ້ອມ virtual ໂດຍການເຮັດວຽກທັງສອງຄໍາສັ່ງນີ້:

$>source bin/activate

(iceberg_playground)$>pip install pyiceberg daft duckdb sqlalchemy pandas

ສໍາລັບຕົວຢ່າງນີ້, PyIceberg ຈະນໍາໃຊ້ SqlCatalog, ເຊິ່ງອັບໂຫລດຂໍ້ມູນຄອມພິວເຕີ Iceberg ໃນຖານຂໍ້ມູນ SQLite ພື້ນຖານ. Iceberg ຍັງສະຫນັບສະຫນູນ catalogues ລວມທັງ REST, Hive, ແລະ AWS Glue.

ເອກະສານການຄຸ້ມຄອງ .pyiceberg.yaml ແມ່ນ prepared ມີ content ນີ້, at the project root:

catalog:

 pyarrowtest:

   uri: sqlite:///pyiceberg_catalog/pyarrow_catalog.db

   warehouse: file:////Users/diptimanraichaudhuri/testing_space/iceberg_playground/dw1

ວິທີການ Iceberg catalogues ໄດ້ຖືກເກັບຮັກສາພາຍໃຕ້ directory pyiceberg_catalog ເປັນເອກະສານ SQLite ແລະໃນອຸປະກອນຂໍ້ມູນ, ເຊິ່ງເກັບຮັກສາຂໍ້ມູນທັງຫມົດແລະ metadata ໃນ directory dw1.

ທັງສອງ directories ນີ້ໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນໃນລະດັບ root ຂອງໂຄງການ. ຊື່ນີ້ແມ່ນ pyarrowtest.

ຫຼັງຈາກນັ້ນ, ການຕິດຕັ້ງ PyIceberg ໄດ້ຖືກຄວບຄຸມໂດຍໃຊ້ script ນີ້:

import os

from pyiceberg.catalog import load_catalog

os.environ["PYICEBERG_HOME"] = os.getcwd()

catalog = load_catalog(name='pyarrowtest')

print(catalog.properties)

ຊອກຫາວິທີທີ່ PyIceberg ດາວນ໌ໂຫລດຊື່ catalogues ຈາກເອກະສານ YAML ແລະສ້າງ database SQLite ພື້ນທີ່ໃນ directory pyiceberg_catalog. ໃນຂະນະທີ່ SQLite ໄດ້ຖືກສົ່ງອອກກັບຜູ້ຕິດຕັ້ງ Python, ມັນບໍ່ຈໍາເປັນຕ້ອງຕິດຕັ້ງ separately.

If the script is run properly, the 'pyarrow_catalog' properties should be displayed in the terminal.

Script ດາວໂຫລດ catalogues ຈາກ .YAML file, ໃນຂະນະທີ່ 'PYICEBEG_HOME' ປະລິມານສະພາບແວດລ້ອມໄດ້ຖືກພິຈາລະນາເປັນ root ຂອງໂຄງການ.

ຫຼັງຈາກນັ້ນ, schema ແມ່ນເພີ່ມຂຶ້ນໂດຍໃຊ້ຊັ້ນ schema ຂອງ PyIceberg. ເນື່ອງຈາກວ່າຕົວຢ່າງນີ້ອັບໂຫລດຂໍ້ມູນຈາກກຸ່ມຂອງ sensors IoT, schema ແມ່ນສ້າງຕັ້ງຂຶ້ນດ້ວຍສາມຊັ້ນແລະປະເພດຂໍ້ມູນທີ່ຕ້ອງການຂອງພວກເຂົາ. field device_id ໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນທັງສອງເປັນ key primary ແລະເປັນ key partition.

ຫຼັງຈາກນັ້ນ, namespace ໄດ້ຖືກສ້າງຂຶ້ນພາຍໃຕ້ການ Iceberg table ກັບ schema. A namespace ແມ່ນການກໍານົດບັນຊີລາຍລະອຽດຂອງທ້ອງຖ່າຍຮູບໃນອຸປະກອນ (ລັກສະນະຂອງອຸປະກອນໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນໃນເວລາທີ່ຄັດເລືອກເອົາເອກະສານ YAML).

ການດາວໂຫລດຂໍ້ມູນທີ່ເລີ່ມຕົ້ນຖືກເຮັດໂດຍໃຊ້ການນໍາໃຊ້ລັກສະນະ in-memory ຂອງ PyArrow, ແລະ sensor_table ໄດ້ຖືກອ່ານໂດຍໃຊ້ວິທີການ PyIceberg scan() ເພື່ອປັບປຸງຂໍ້ມູນໃນ frame data pandas.

import os

from pyiceberg.catalog import load_catalog

from pyiceberg.schema import Schema

from pyiceberg.types import (NestedField,

                             StringType, FloatType)

from pyiceberg.partitioning import PartitionSpec, PartitionField

from pyiceberg.transforms import IdentityTransform

import pyarrow as pa


os.environ["PYICEBERG_HOME"] = os.getcwd()

catalog = load_catalog(name='pyarrowtest')

print(catalog.properties)

# Define the schema

schema = Schema(

    NestedField(1, "device_id", StringType(), required=True),

    NestedField(2, "ampere_hour", FloatType(), required=True),

    NestedField(3, "device_make", StringType(), required=True),

    identifier_field_ids=[1]  # 'device_id' is the primary key

)

# Define the partition spec - device_id as partition key

partition_spec = PartitionSpec(

    PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="device_id")

)

# Create a namespace and an iceberg table

catalog.create_namespace_if_not_exists('sensor_ns')

sensor_table = catalog.create_table_if_not_exists(

    identifier='sensor_ns.sensor_data',

    schema=schema,

    partition_spec=partition_spec

)

# Insert initial data

initial_data = pa.table([

    pa.array(['b8:27:eb:bf:9d:51', '00:0f:00:70:91:0a', '1c:bf:ce:15:ec:4d']),

    pa.array([21.43, 17.86, 31.65]),

    pa.array(['ABB', 'Honeywell', 'Siemens'])

], schema=schema.as_arrow())

# Write initial data

sensor_table.overwrite(initial_data)

# Print a Pandas dataframe representation of the table data

print("\nInsert operation completed")

print(sensor_table.scan().to_pandas())

ຖ້າຫາກວ່າ scripts ທີ່ຜ່ານມາຖືກປະຕິບັດຢ່າງງ່າຍດາຍ, ຜົນນີ້ຈະສະແດງຢູ່ໃນ terminal:

ໃນຂະນະທີ່ insert ໄດ້ຖືກປັບປຸງຢ່າງງ່າຍດາຍ, ຜະລິດຕະພັນຂອງ Iceberg ສາມາດໄດ້ຮັບການຢັ້ງຢືນ:

1. Catalogues: ມັນເປັນເອກະສານ SQLite pyarrow_catalog.db, ເຊິ່ງຈະໄດ້ຮັບການຄວບຄຸມຫຼັງຈາກນັ້ນໃນບົດຄວາມນີ້.
2. Metadata: ໃນ directory 'metadata' ທີ່ມີເອກະສານ metadata ໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນ, ເຊິ່ງເປັນສິ່ງທີ່ສໍາຄັນສໍາລັບການປະຕິບັດການ Create, Read, Update, Delete (CRUD). ສອງເອກະສານ metadata JSON ໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນ, ຫນຶ່ງໃນເວລາທີ່ທັນສະໄຫມໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນ, ແລະອື່ນໆຫຼັງຈາກ insert ຂອງຂໍ້ມູນຄັ້ງທໍາອິດ. ເອກະສານ 'snap-*.avro' ເປັນລາຍລະອຽດ manifesto, ແລະ ເອກະສານ manifesto ເປັນເອກະສານ .avro ອື່ນໆ.
3. ຂໍ້ມູນ: ຂໍ້ມູນໄດ້ຖືກຂຽນໃນຮູບແບບ .PARQUET, ມີ device_id ເປັນ key partition. ເນື່ອງຈາກວ່າພວກເຮົາມີສາມອຸປະກອນທີ່ແຕກຕ່າງກັນ, ຂໍ້ມູນໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນໃນສາມ directories. The Iceberg table 'sensor_data' is created with the namespace 'sensor_ns.db' in the warehouse 'dw1'. These data fields are created in the 'data' directory of the 'sensor_data' table.

ການສອບເສັງ PyIceberg ສາມາດຖືກນໍາໃຊ້ເພື່ອກວດກາບັນຫາ. ການສອບເສັງສ່ວນໃຫຍ່ທີ່ນໍາໃຊ້ສໍາລັບການກວດກາແມ່ນ: StartsWith, EqualTo, GreaterThan, And, Or, ແລະອື່ນໆ.

from pyiceberg.expressions import StartsWith, EqualTo

# Filter columns

print("\nFilter records with device_make == ABB ")

print(sensor_table.scan(row_filter=EqualTo('device_make', 'ABB')).to_pandas())

PyIceberg ສະຫນັບສະຫນູນການເຮັດວຽກ UPSERT ເຊັ່ນດຽວກັນ. ລະຫັດຕົວຢ່າງທີ່ຜ່ານມາໄດ້ປັບປຸງ device_make ທີ່ມີຢູ່ໃນຫນຶ່ງຂອງ sensors ຈາກ ‘Siemens’ ກັບ ‘Kepware’.

# Create an UPSERT batch of Arrow records where one fot he device_make is changed

upsert_data = pa.table([

    pa.array(['b8:27:eb:bf:9d:51', '00:0f:00:70:91:0a', '1c:bf:ce:15:ec:4d']),

    pa.array([21.43, 17.86, 31.65]),

    pa.array(['ABB', 'Honeywell', 'Kepware'])

], schema=schema.as_arrow())

# UPSERT changed data

try:

    join_columns = ["device_id"]

    upsert_result = sensor_table.upsert(upsert_data.select(["device_id", "ampere_hour", "device_make"]))

except Exception as e:

    print(e)

print("\nUpsert operation completed")

print(sensor_table.scan().to_pandas())

ໃນວິທີທີ່ເຫມາະສົມ, ການປະຕິບັດ DELETE ຍັງຖືກສະຫນັບສະຫນູນ:

# Delete row

sensor_table.delete(delete_filter=EqualTo('device_id', '1c:bf:ce:15:ec:4d'))

print("\n After Delete")

print(sensor_table.scan().to_pandas())

ການແກ້ໄຂໃນອຸປະກອນໃດກໍ່ຕາມເປັນການປະຕິບັດທີ່ແຕກຕ່າງກັນ, ແລະ Iceberg ແມ່ນບໍ່ແມ່ນອະນຸຍາດ. ການແກ້ໄຂໃນອຸປະກອນ Iceberg ແມ່ນຖືກຄັດເລືອກໂດຍສອງສະຖານທີ່: Copy-on-Write (CoW) ແລະ Merge-on-Read (MoR). ການແກ້ໄຂຍັງສ້າງເອກະສານແກ້ໄຂສໍາລັບສະຖານທີ່ MoR.

PyIceberg ໃນປັດຈຸບັນສະຫນັບສະຫນູນການດາວໂຫລດ MOR, ແຕ່ມີຂະຫນາດນ້ອຍ. ໃນຂະນະທີ່ PyIceberg ສະຫນັບສະຫນູນຄວາມສາມາດໃນການດາວໂຫລດຊຸດ, ມັນ primarily ການນໍາໃຊ້ການດາວໂຫລດ CoW ໂດຍປົກກະຕິ, ເຊິ່ງເປັນການດາວໂຫລດຟອມພິວເຕີໂດຍບໍ່ມີການສ້າງຟອມພິວເຕີທີ່ດາວໂຫລດ. ຢ່າງໃດກໍຕາມ, ມີການເຮັດວຽກໃນປັດຈຸບັນເພື່ອປັບປຸງ MoR ຂອງ PyIceberg ເພື່ອສະຫນັບສະຫນູນການດາວໂຫລດຄຸນນະພາບແລະເຮັດໃຫ້ມັນມີປະສິດທິພາບຫຼາຍສໍາລັບການປັບປຸງຂະຫນາດນ້ອຍ.

ໃນຕອນທ້າຍ, ພວກເຮົາຈະນໍາໃຊ້ DuckDB ເພື່ອຊອກຫາ catalogu Iceberg SQLite, ທີ່ຖືກເກັບຮັກສາໃນເອກະສານ pyarrow_catalog.db. ຄໍາຖາມທີ່ຜ່ານມາແມ່ນປະຕິບັດໃນ terminal VS Code:

duckdb -ui pyiceberg_catalog/pyarrow_catalog.db

ນີ້ຈະເປີດ window browser at port 4213 (default), ໃນເວລາທີ່ການຊອກຫາ SQL ສາມາດໄດ້ຮັບການປະຕິບັດໃນ Iceberg catalogues, ເຊັ່ນດຽວກັນ:

ນີ້ແມ່ນວິທີທີ່ງ່າຍແລະງ່າຍທີ່ຈະຊອກຫາຄວາມຮູ້ຈາກ catalogues SQL

ດາວນ໌ໂຫລດ PyIceberg

ສໍາລັບບໍລິສັດທີ່ມີຂະຫນາດນ້ອຍກວ່າ terabytes, PyIceberg ແລະ PyArrow ແມ່ນການເລືອກທີ່ດີທີ່ສຸດສໍາລັບການດໍາເນີນການຊອບແວ interactive ໃນ Python. ການນໍາໃຊ້ການເຊື່ອມຕໍ່ນີ້ສາມາດຫຼຸດຜ່ອນເວລາທີ່ສໍາຄັນເພື່ອຜະລິດຄວາມຮູ້ສຶກສໍາລັບຊອບແວຂະຫນາດນ້ອຍຫຼືຂະຫນາດກາງ.

ວິສະວະກອນ Data ສາມາດເລີ່ມຕົ້ນການຢັ້ງຢືນ PyIceberg, ທີ່ຖືກປິ່ນປົວແລະໄດ້ຮັບການປິ່ນປົວໃນມື້ນີ້. ນອກເຫນືອຈາກການພາສາລາວຫນ້ານີ້ມີຕົວຢ່າງທີ່ດີຂອງທັງຫມົດ API PyIceberg.

ລະຫັດ QR