예니의 코딩일기

실습 fake_people 토픽의 title top 10 계산하기 spark.jars.packages 설정 spark-defaults.conf 파일 수정 SparkSession 생성시 config 지정 spark-submit 실행시 --packages 옵션 사용 # kafka_source_streaming.py from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, expr from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType, IntegerType, ArrayType if __name__ == "__m..

강의 ksqlDB "ksqlDB"는 REST API 혹은 ksql 클라이언트 툴을 이용하여 Topic을 테이블처러 SQL로 조작할 수 있는 방법 입니다. docker ps docker exec -it ContainerID sh ksql CREATE STREAM my_stream (id STRING, name STRING, title STRING) with (kafka_topic='fake_peaple', value_format='JSON'); SELECT * FROM my_stream;

실습 랜덤하게 사람 정보 생성 및 저장하는 KafkaProducer 구현 Faker 모듈 사용 pydantic 모듈의 BaseModel 사용 Topic 생성 후 추가 pip install faker pip install pydantic # person.py from pydantic import BaseModel class Person(BaseModel): id: str name: str title: str # fake_person_producer.py import uuid import json from typing import List from person import Person from faker import Faker from kafka.admin import NewTopic from kafka.e..

강의 Kafka 프로그래밍 언어 Java Apache Kafka Java Client Spring Kafka Python Confluent Kafka Python Kafka-Python pip install kafka-python .NET Confluent Kafka .NET Go Sarama Node.js node-rdkafka kafka-node Kafka 기본 프로그래밍 Producer 생성 from time import sleep from json import dumps from kafka import KafkaProducer producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda x: dumps..

강의 Kafka 설치 Docker Compose 사영 Github repo 사용 : https://github.com/conduktor/kafka-stack-docker-compose full-stack.yml zk-single-kafka-single.yml zk-single-kafka-multiple.yml zk-multiple-kafka-single.yml zk-multiple-kafka-multiple.yml Kafka 웹 UI localhost:8080 ID : admin@admin.io / Password : admin git clone https://github.com/conduktor/kafka-stack-docker-compose cd kafka-stack-docker-compose doc..

강의 Kafka Connect "Kafka Connect"는 Kafka 위에 생성된 중앙 집중 데이터 허브 입니다. Broker 일부 혹은 별도 서버로 구성 → 별도 오픈소스 프로젝트 데이터 버스 혹은 메시지 버스 모드 Standalone 모드 : 개발 혹은 테스트 Distributed 모드 데이터베이스, 파일 시스템, 키-값 저장소, 검색 인덱스 등 저장 데이터 소스 및 데이터 싱크 Kafka Schema Registry "Kafka Schema Registry"는 Topic 메시지 데이터에 대한 스키마를 관리 및 검증합니다. 포맷 변천(Schema Evolution) 지원 : Schema ID 혹은 버전 사용 포맷 변경 종류 Forward Compatibility : Producer 변경 후 Cons..