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

01-Producer.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notebook de ejemplo para mostrar cómo producir-consumir mensajes de kafka desde python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting kafka-python-ng\n",
      "  Downloading kafka_python_ng-2.2.2-py2.py3-none-any.whl.metadata (9.2 kB)\n",
      "Downloading kafka_python_ng-2.2.2-py2.py3-none-any.whl (232 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m232.4/232.4 kB\u001b[0m \u001b[31m1.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0m\n",
      "\u001b[?25hInstalling collected packages: kafka-python-ng\n",
      "Successfully installed kafka-python-ng-2.2.2\n",
      "Note: you may need to restart the kernel to use updated packages.\n"
     ]
    }
   ],
   "source": [
    "%pip install kafka-python-ng"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import KafkaProducer from Kafka library and create a producer object\n",
    "from kafka import KafkaProducer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "kakfa_producer = KafkaProducer(bootstrap_servers='kafka:9092')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i in range(10):\n",
    "    kakfa_producer.send('test', f'New messages {i}'.encode(\"utf-8\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<kafka.producer.future.FutureRecordMetadata at 0x7f304c295880>"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kakfa_producer.send('test2', b'Hello, World2!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<kafka.producer.future.FutureRecordMetadata at 0x7f304c295010>"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kakfa_producer.send('test', b'exit')"
   ]
  },
  {
   "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.12.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}

02-consumer.ipynb

03-spark.ipynb

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