Ejemplo introducción a Kafka - christian-ariza.net

01-Producer.ipynb

02-consumer.ipynb

03-spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "\n",
    "# Create a SparkSession configured to read from kafka\n",
    "spark = (SparkSession.builder.appName(\"kafka\").master(\"spark://spark-master:7077\")\n",
    "         .config(\"spark.jars.packages\", \"org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.2,org.apache.hadoop:hadoop-aws:3.3.4\")\n",
    "            .config(\"spark.hadoop.fs.s3a.access.key\", \"minio\")\n",
    "            .config(\"spark.hadoop.fs.s3a.secret.key\", \"minio2024\")\n",
    "            .config(\"spark.hadoop.fs.s3a.endpoint\", \"http://minio-server:9000\")\n",
    "            .config(\"spark.driver.blockManager.port\", \"7777\")\n",
    "            .config(\"spark.driver.port\", \"38889\")\n",
    "            .config(\"spark.driver.host\", \"127.0.0.1\")\n",
    "            .config(\"spark.driver.bindAddress\", \"0.0.0.0\")\n",
    "            .config(\"spark.hadoop.fs.s3a.path.style.access\", \"true\")\n",
    "            .config( \"spark.hadoop.fs.s3a.aws.credentials.provider\", \"org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider\")\n",
    "            .config(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
    "\n",
    ".getOrCreate())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kafka_df = (spark.readStream.format(\"kafka\")\n",
    "            .option(\"kafka.bootstrap.servers\", \"kafka:9092\").option(\"subscribe\", \"test\")\n",
    "            .option(\"startingOffsets\", \"latest\").load())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "query1 = (kafka_df\n",
    "          .selectExpr(\"CAST(key AS STRING) as key\", \"CAST(value AS STRING) as value\")\n",
    "          .writeStream\n",
    "          .format(\"console\")\n",
    "          .trigger(once=True)\n",
    "          .start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = (kafka_df\n",
    "          .selectExpr(\"CAST(key AS STRING) as key\", \"CAST(value AS STRING) as value\")\n",
    "          .writeStream\n",
    "          .queryName(\"toCSV\")\n",
    "          .format(\"csv\")\n",
    "          .option(\"path\", \"s3a://test-bucket/test_path\")\n",
    "          .option(\"checkpointLocation\", \"s3a://test-bucket/checkpoint\")\n",
    "         .trigger(processingTime=\"1 minute\")\n",
    "           .start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for s in spark.streams.active:\n",
    "    s.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}

docker-compose.yml

version: '3'
services:
  kafka:
    image: confluentinc/cp-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    depends_on:
      - zookeeper

  zookeeper:
    image: confluentinc/cp-zookeeper
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181
    ports:
      - "2181:2181"
  spark-executor:
    image: spark:3.5.1-scala2.12-java11-ubuntu
    environment:
      SPARK_MASTER_HOST: spark-master
      SPARK_MASTER_PORT: 7077
    command: spark-class org.apache.spark.deploy.worker.Worker spark://spark-master:7077
    volumes:
      - ./shared_folder:/workspace/data
    deploy:
      replicas: 2
    depends_on:
      - spark-master
  spark-master:
    image: spark:3.5.1-scala2.12-java11-ubuntu
    environment:
      SPARK_MASTER_HOST: spark-master
      SPARK_MASTER_PORT: 7077
      SPARK_LOCAL_IP: spark-master
    command: spark-class org.apache.spark.deploy.master.Master
    volumes:
      - ./shared_folder:/workspace/data
  spark-notebook:
    build:
      context: .
      dockerfile_inline: |
        FROM quay.io/jupyter/pyspark-notebook:spark-3.5.1
        #set password for jupyter
        RUN echo '{"IdentityProvider": {"hashed_password": "argon2:$$argon2id$$v=19$$m=10240,t=10,p=8$$zUYc32oQmbROa0YxSdntdw$$5hOMyxMMdml9/pM1Jc8A1GNMhi1d3cEEZiBW3KjJhCY"}}' >> /home/jovyan/.jupyter/jupyter_server_config.json && \
        chmod 600 /home/jovyan/.jupyter/jupyter_server_config.json
    ports:
      - "4040:4040"
      - "8888:8888"
      - "38889:38889"
      - "7777:7777"
    volumes:
      - ./:/home/jovyan/work
  minio-server:
    image: quay.io/minio/minio
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: minio2024