pysparkとicebergでデータを作る。あとS3にアップロードする。

インストール

pip install pyspark==3.5.3
pip install "pyspark[sql]" 
# Iceberg Spark Runtime (Spark 3.5 用)
pip install iceberg-spark-runtime-3.5_2.12:1.6.1
# または最新版を指定
pip install pyspark pandas pyarrow

javaを入れる

apt update && apt install openjdk-17-jdk -y && \
echo 'export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64' >> /etc/profile && \
source /etc/profile && java -version

from pyspark.sql import SparkSession
import datetime

# Spark + Iceberg 起動（JAR自動ダウンロード付き！）
spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("Iceberg 最終完全版")
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.local.type", "hadoop")
    .config("spark.sql.catalog.local.warehouse", "/tmp/iceberg_warehouse")
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1")  # これが鍵！ 初回だけダウンロード
    .getOrCreate()
)

print("Spark + Iceberg 起動中...（初回はJARダウンロードで少し時間がかかります）")

# データベース作成（エラー出ないはず！）
spark.sql("CREATE DATABASE IF NOT EXISTS local.db")
print("データベース確認")
spark.sql("SHOW DATABASES IN local").show()

# サンプルデータ作成（日本人ユーザー10人）
data = [
    (1, "田中太郎", 30, "male", datetime.date(2025, 1, 1)),
    (2, "鈴木花子", 25, "female", datetime.date(2025, 1, 2)),
    (3, "佐藤次郎", 35, "male", datetime.date(2025, 1, 3)),
    (4, "高橋美咲", 28, "female", datetime.date(2025, 1, 4)),
    (5, "渡辺健太", 40, "male", datetime.date(2025, 1, 5)),
    (6, "山本あかり", 22, "female", datetime.date(2025, 1, 6)),
    (7, "中村大輔", 33, "male", datetime.date(2025, 1, 7)),
    (8, "小林彩花", 27, "female", datetime.date(2025, 1, 8)),
    (9, "松本翔太", 31, "male", datetime.date(2025, 1, 9)),
    (10, "加藤真由", 29, "female", datetime.date(2025, 1, 10)),
]

df = spark.createDataFrame(data, "id LONG, name STRING, age INT, gender STRING, dt DATE")

# Icebergテーブル作成＆データ挿入
df.writeTo("local.db.users") \
  .partitionedBy("dt") \
  .createOrReplace()

print("Icebergテーブル作成＆データ挿入完了！！")

# データ確認
print("\n【全件表示】")
spark.table("local.db.users").show(truncate=False)

print("\n【女性だけ】")
spark.table("local.db.users").filter("gender = 'female'").show()

print("\n【スナップショット履歴（Time Travel可能！）】")
spark.sql("SELECT snapshot_id, committed_at, operation FROM local.db.users.snapshots").show(truncate=False)

print("\nデータフォルダ： /tmp/iceberg_warehouse/db/users/data/ （ls で確認！）")

spark.stop()

S3にアップロード

aws s3 mb s3://my-iceberg-warehouse-2025 --region ap-northeast-1

from pyspark.sql import SparkSession
import datetime

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("Iceberg S3 + Glue 本番環境")
    # これがS3+Glueの最重要設定！
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.glue_catalog.warehouse", "s3a://my-iceberg-warehouse-2025/warehouse/")
    .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
    .config("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
    # JAR自動ダウンロード（初回だけ時間かかる）
    .config("spark.jars.packages",
            "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,"
            "org.apache.hadoop:hadoop-aws:3.3.6,"
            "software.amazon.awssdk:bundle:2.20.160,"
            "software.amazon.awssdk:url-connection-client:2.20.160")
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
    # あなたのAWS認証（3パターンどれか選んで！）
    # パターンA: AWS CLIで login済み → 何も書かなくてOK！
    # パターンB: 環境変数
    # .config("spark.hadoop.fs.s3a.access.key", "AKIXXXXXXXXXXXXXXXXX")
    # .config("spark.hadoop.fs.s3a.secret.key", "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")
    # パターンC: IAMロール（EMRやEC2の場合）
    .getOrCreate()
)

print("S3 + Glue Catalog で Iceberg 起動成功！！")

# Glueにデータベース自動作成
spark.sql("CREATE DATABASE IF NOT EXISTS glue_catalog.iceberg_db")
print("Glueデータベース一覧")
spark.sql("SHOW DATABASES IN glue_catalog").show()

# サンプルデータ
data = [(i, f"ユーザー{i}", 20+i, datetime.date(2025,12,9)) for i in range(1,6)]
df = spark.createDataFrame(data, "id LONG, name STRING, age INT, dt DATE")

# S3にIcebergテーブル作成！
df.writeTo("glue_catalog.iceberg_db.users_s3") \
  .partitionedBy("dt") \
  .tableProperty("format-version", "2") \
  .createOrReplace()

print("S3にIcebergテーブル作成完了！！")
spark.table("glue_catalog.iceberg_db.users_s3").show()

print("データはここに保存されました →")
print("s3a://my-iceberg-warehouse-2025/warehouse/iceberg_db.db/users_s3/")

spark.stop()