G\u00fcn\u00fcm\u00fcz veri platformlar\u0131nda g\u00fcncel ihtiya\u00e7lardan biri, veriyi g\u00fcvenilir \u015fekilde saklamak, ger\u00e7ek zamanl\u0131 de\u011fi\u015fiklikleri i\u015flemek ve analitik i\u00e7in tutarl\u0131 tablolar sunmak. Bu noktada a\u00e7\u0131k kaynak \u00e7\u00f6z\u00fcmler b\u00fct\u00e7e a\u00e7\u0131s\u0131ndan sundu\u011fu rahatl\u0131k ile g\u00fcndeme geliyor. Bu yaz\u0131da MinIO, Apache Iceberg, Spark, Debezium, Kafka, Airflow ve Trino gibi a\u00e7\u0131k kaynakl\u0131 ara\u00e7lar\u0131 bir araya getirerek u\u00e7tan uca bir veri g\u00f6l\u00fc senaryosu kurgulayaca\u011f\u0131m.<\/p>\n\n\n\n
Iceberg, klasik Hive tabanl\u0131 tablolara g\u00f6re bir\u00e7ok avantaj sa\u011flar:<\/p>\n\n\n\n
K\u0131sacas\u0131 uzun vadeli, g\u00fcvenilir ve analitik odakl\u0131 bir veri g\u00f6l\u00fc<\/strong> i\u00e7in Iceberg en mant\u0131kl\u0131 tercihlerden biridir.<\/p>\n\n\n\n MSSQL \u2192 Debezium \u2192 Kafka \u2192 Spark Structured Streaming \u2192 Iceberg<\/strong><\/p>\n\n\n\n Debezium, veritaban\u0131ndaki de\u011fi\u015fiklikleri Kafka\u2019ya CDC event\u2019leri olarak aktar\u0131r.<\/p>\n\n\n\n Connector config \u00f6rne\u011fi (MSSQL i\u00e7in):<\/strong><\/p>\n\n\n\n Kafka\u2019ya d\u00fc\u015fen veriler genelde \u015fu formatta olur:<\/p>\n\n\n\n op: i\u015flem tipi (c=create, u=update, d=delete).<\/p>\n\n\n\n Structured Streaming job\u2019u, Kafka topic\u2019ini okuyup Iceberg tablosuna yazar.<\/p>\n\n\n\n Bu job s\u00fcrekli \u00e7al\u0131\u015farak ger\u00e7ek zamanl\u0131 g\u00fcncellenmi\u015f bir Iceberg tablosu<\/strong> sa\u011flar.<\/p>\n\n\n\n MSSQL \u2192 Debezium \u2192 Kafka \u2192 Airflow DAG \u2192 Spark Batch \u2192 Iceberg<\/strong><\/p>\n\n\n\n Spark Batch Job (kafka_batch_to_iceberg.py)<\/strong><\/p>\n\n\n\n Burada batch olarak Kafka\u2019dan al\u0131nan veri Iceberg\u2019e yaz\u0131l\u0131r. Saatlik, g\u00fcnl\u00fck veya custom aral\u0131klarla \u00e7al\u0131\u015ft\u0131r\u0131labilir.<\/p>\n\n\n\n MSSQL \u2192 Debezium \u2192 Kafka \u2192 Kafka Connect Iceberg Sink \u2192 Iceberg<\/strong><\/p>\n\n\n\n Avantaj: Spark ba\u011f\u0131ml\u0131l\u0131\u011f\u0131 yok, sade mimari. Bu konfig\u00fcrasyon MSSQL\u2019de CDC aktif tablolar\u0131 dinler ve Kafka\u2019da \u015fu formatta topic olu\u015fturur:<\/p>\n\n\n\n mssql_server1.dbo.users<\/em><\/p>\n\n\n\n mssql_server1.dbo.orders<\/em><\/p>\n\n\n\n Bu ayarlarda:<\/p>\n\n\n\n Airflow, Kafka topic\u2019ini dinleyerek yeni veri geldi\u011finde Spark job tetikleyebilir.<\/p>\n\n\n\n \u00d6rnek DAG<\/strong><\/p>\n\n\n\n Burada Spark job\u2019u ancak Kafka\u2019ya veri d\u00fc\u015ft\u00fc\u011f\u00fcnde \u00e7al\u0131\u015f\u0131r.<\/p>\n\n\n\n MSSQL Taraf\u0131nda CDC Ayarlar\u0131<\/strong><\/p>\n\n\n\n etc\/catalog\/iceberg.properties:<\/em><\/p>\n\n\n\n Sorgu \u00f6rne\u011fi:<\/p>\n\n\n\n \u00d6zetle, bu yaz\u0131da MSSQL \u2192 Iceberg entegrasyonunu birden fazla y\u00f6ntem \u00fczerinden ele alarak farkl\u0131 mimari tercihlere g\u00f6re uygulanabilecek temel yakla\u015f\u0131mlara yer verdim.<\/p>\n\n\n\n Buradaki de\u011ferlendirmelerin, kurumsal veri platformlar\u0131n\u0131 modernle\u015ftirmek ve analitik ihtiya\u00e7lar i\u00e7in daha esnek bir altyap\u0131 olu\u015fturmak isteyen ekipler i\u00e7in yol g\u00f6sterici olaca\u011f\u0131n\u0131 d\u00fc\u015f\u00fcn\u00fcyorum.<\/a><\/p>\n\n\n\nSenaryolar<\/h2>\n\n\n\n
Senaryo A \u2014 Spark Structured Streaming<\/h2>\n\n\n\n
Debezium Kurulumu ve Kafka Entegrasyonu<\/h2>\n\n\n\n
{
\"name\": \"mssql-connector\",
\"config\": {
\"connector.class\": \"io.debezium.connector.sqlserver.SqlServerConnector\",
\"database.hostname\": \"mssql_host\",
\"database.port\": \"1433\",
\"database.user\": \"user\",
\"database.password\": \"pass\",
\"database.dbname\": \"mydb\",
\"database.server.name\": \"mssql_server\",
\"table.include.list\": \"dbo.customers\",
\"database.history.kafka.bootstrap.servers\": \"kafka:9092\",
\"database.history.kafka.topic\": \"schema-changes.mssql\"
}
}<\/pre>\n\n\n\n{
\"before\": {\"id\": 1, \"name\": \"Ali\"},
\"after\": {\"id\": 1, \"name\": \"Ali Veli\"},
\"op\": \"u\",
\"ts_ms\": 1690000000000
}<\/pre>\n\n\n\nSpark Structured Streaming \u0130\u015flemi<\/h2>\n\n\n\n
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType<\/pre>\n\n\n\nspark = (SparkSession.builder
.appName(\"cdc-to-iceberg\")
.config(\"spark.sql.catalog.my_catalog\", \"org.apache.iceberg.spark.SparkCatalog\")
.config(\"spark.sql.catalog.my_catalog.type\", \"hadoop\")
.config(\"spark.sql.catalog.my_catalog.warehouse\", \"s3a:\/\/warehouse\/iceberg\")
.getOrCreate())schema = StructType([
StructField(\"before\", StringType()),
StructField(\"after\", StringType()),
StructField(\"op\", StringType()),
StructField(\"ts_ms\", StringType())
])kafka_df = (spark.readStream
.format(\"kafka\")
.option(\"kafka.bootstrap.servers\", \"kafka:9092\")
.option(\"subscribe\", \"mssql_server.dbo.customers\")
.load())parsed_df = kafka_df.selectExpr(\"CAST(value AS STRING) as json\") \\
.select(from_json(col(\"json\"), schema).alias(\"data\"))# \u00d6rne\u011fin sadece insert\/update olan kay\u0131tlar\u0131 after\u2019dan al\u0131p yaz\u0131yoruz
final_df = parsed_df.select(\"data.after.*\")(final_df.writeStream
.format(\"iceberg\")
.option(\"checkpointLocation\", \"s3a:\/\/chkpt\/customers\/\")
.toTable(\"my_catalog.db.customers\"))<\/pre>\n\n\n\nSenaryo B \u2014 Kafka + Airflow + Spark Batch<\/h2>\n\n\n\n
Airflow DAG (Kafka \u2192 Spark Batch)<\/h2>\n\n\n\n
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.utils.dates import days_ago<\/pre>\n\n\n\ndefault_args = {\"owner\": \"airflow\", \"start_date\": days_ago(1)}dag = DAG(
dag_id=\"kafka_to_iceberg_batch\",
default_args=default_args,
schedule_interval=\"@hourly\",
catchup=False
)
spark_job = SparkSubmitOperator(
task_id=\"spark_batch_load\",
application=\"\/opt\/spark_jobs\/kafka_batch_to_iceberg.py\",
conn_id=\"spark_default\",
dag=dag
)<\/pre>\n\n\n\nkafka_df = (spark.read
.format(\"kafka\")
.option(\"kafka.bootstrap.servers\", \"kafka:9092\")
.option(\"subscribe\", \"mssql_server.dbo.customers\")
.load())<\/pre>\n\n\n\nparsed_df = kafka_df.selectExpr(\"CAST(value AS STRING)\")
parsed_df.writeTo(\"my_catalog.db.customers\").append()<\/pre>\n\n\n\nSenaryo C \u2014 Kafka Connect \u2192 Iceberg Sink (Spark\u2019s\u0131z)<\/h2>\n\n\n\n
Kullan\u0131m Alan\u0131: Daha basit kurulumlar.<\/p>\n\n\n\nDebezium MSSQL Source Connector (connector-mssql.json)<\/h3>\n\n\n\n
{
\"name\": \"mssql-debezium-connector\",
\"config\": {
\"connector.class\": \"io.debezium.connector.sqlserver.SqlServerConnector\",
\"tasks.max\": \"1\",
\"database.hostname\": \"mssql-host\",
\"database.port\": \"1433\",
\"database.user\": \"sa\",
\"database.password\": \"StrongPassword!\",
\"database.dbname\": \"mydb\",<\/pre>\n\n\n\n \"database.server.name\": \"mssql_server1\",
\"table.include.list\": \"dbo.users,dbo.orders\",
\"database.encrypt\": \"false\",
\"snapshot.mode\": \"initial\",
\"database.history.kafka.bootstrap.servers\": \"kafka:9092\",
\"database.history.kafka.topic\": \"schema-changes.mydb\",
\"include.schema.changes\": \"true\",
\"tombstones.on.delete\": \"false\",
\"decimal.handling.mode\": \"string\",
\"time.precision.mode\": \"connect\"
}
}<\/pre>\n\n\n\nIceberg Sink Connector (connector-iceberg.json)<\/h3>\n\n\n\n
{
\"name\": \"iceberg-sink-connector\",
\"config\": {
\"connector.class\": \"org.apache.iceberg.connect.IcebergSinkConnector\",
\"tasks.max\": \"1\",<\/pre>\n\n\n\n \"topics\": \"mssql_server1.dbo.users\",
\"iceberg.catalog.type\": \"hadoop\",
\"iceberg.catalog.warehouse\": \"s3a:\/\/datalake\/warehouse\",
\"iceberg.catalog.hadoop.fs.s3a.endpoint\": \"http:\/\/minio:9000\",
\"iceberg.catalog.hadoop.fs.s3a.access.key\": \"minioadmin\",
\"iceberg.catalog.hadoop.fs.s3a.secret.key\": \"minioadmin\",
\"iceberg.catalog.hadoop.fs.s3a.path.style.access\": \"true\",
\"iceberg.catalog.name\": \"hive_prod\",
\"iceberg.table.auto-create\": \"true\",
\"iceberg.table.namespace\": \"db1\",
\"iceberg.table.name\": \"users\",
\"key.converter\": \"org.apache.kafka.connect.json.JsonConverter\",
\"value.converter\": \"org.apache.kafka.connect.json.JsonConverter\",
\"value.converter.schemas.enable\": \"true\"
}
}<\/pre>\n\n\n\n\n
Senaryo D \u2014 Kafka Sensor DAG<\/h2>\n\n\n\n
from airflow import DAG
from airflow.providers.apache.kafka.sensors.kafka import KafkaSensor
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.utils.dates import days_ago<\/pre>\n\n\n\ndefault_args = {\"owner\": \"airflow\", \"start_date\": days_ago(1)}dag = DAG(\"kafka_sensor_to_spark\", default_args=default_args, schedule_interval=None)wait_for_kafka = KafkaSensor(
task_id=\"wait_for_kafka_msg\",
topic=\"mssql_server.dbo.customers\",
kafka_conn_id=\"kafka_default\",
dag=dag
)
spark_task = SparkSubmitOperator(
task_id=\"spark_job\",
application=\"\/opt\/spark_jobs\/kafka_to_iceberg.py\",
conn_id=\"spark_default\",
dag=dag
)wait_for_kafka >> spark_task<\/pre>\n\n\n\nALTER DATABASE TestDB SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON);
ALTER TABLE dbo.Customers ENABLE CHANGE_TRACKING;<\/pre>\n\n\n\nIceberg Partitioning Stratejisi<\/h2>\n\n\n\n
CREATE TABLE my_catalog.db.customers (
id BIGINT,
name STRING,
country STRING,
updated_at TIMESTAMP
) USING iceberg
PARTITIONED BY (days(updated_at));<\/pre>\n\n\n\nTrino ile Sorgulama<\/h2>\n\n\n\n
connector.name=iceberg
warehouse=s3a:\/\/datalake\/warehouse<\/pre>\n\n\n\nSELECT country, COUNT(*) FROM iceberg.db.customers
WHERE updated_at >= DATE '2025\u201301\u201301' GROUP BY country;<\/pre>\n\n\n\nIncremental ve Full SQL Load \u00d6rnekleri<\/h2>\n\n\n\n
Full Load (Overwrite)<\/h2>\n\n\n\n
df = spark.read.format(\"jdbc\").option(\"url\", jdbc_url).option(\"dbtable\", \"dbo.customers\").load()
df.writeTo(\"my_catalog.db.customers\").overwritePartitions()<\/pre>\n\n\n\nIncremental Load (Append)<\/h2>\n\n\n\n
new_df = spark.read.format(\"jdbc\").option(\"url\", jdbc_url).option(\"dbtable\", \"dbo.customers_delta\").load()
new_df.writeTo(\"my_catalog.db.customers\").append()<\/pre>\n\n\n\nSQL API ile<\/h2>\n\n\n\n
INSERT INTO my_catalog.db.customers SELECT * FROM temp_view;<\/pre>\n\n\n\n