예니의 코딩일기

실습 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..

강의 Spark Streaming "Spark Streaming"는 실시간 데이터 스트림 처리를 위한 Spark API 입니다. 데이터 소스 : Kafk, Kinesis, Flume, TCP 소켓 고급 함수 사용 : Join, Map, Reduce, Window 웹 UI : localhost:4040 동작 방식 마이크로 배치로 데이터 처리 루프 새로운 데이터를 이전 데이터에 Merge 혹은 Replace 데이터 위치 관리 실패시 Fault Tolerance 및 데이터 재처리 관리 최종 결과 스트림 일괄 생성 : Spark Engine DStream RDD 기반 Spark SQL 엔진의 최적화 기능 사용 불가 이벤트 발생 시간 기반 처리 불가 Structured Streaming DataFrame 기반 C..

실습 랜덤하게 사람 정보 생성 및 저장하는 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..

강의 카탈로그 "카탈로그"는 테이블과 뷰에 관한 메타 데이터를 관리합니다. 메모리 기반 카탈로그 기본 Hive 호환 카탈로그 제공 : Persistent 데이터베이스를 통해 테이블을 폴더 구조로 관리 테이블 종류 메모리 기반 테이블/뷰 임시 테이블 스토리지 기반 테이블 HDFS Parquet 포맷 사용 Hive 호환 메타스토어 사용 테이블 종류 Managed Table : 실제 데이터 및 메타 데이터 관리 Unmanaged(External) Table : 메타 데이터 관리 Hive 메타 스토어 구현 스토리지 기반 카탈로그 SparkSession 생성시 enableHiveSupport() 호출 default 데이터베이스 생성 Managed Table 테이블 생성 daaframe.saveAsTable(tab..

실습 실습 3. 사용자별로 처음 채널과 마지막 채널 알아내기 테이블을 데이터프레임으로 로딩 데이터프레임별 테이블 이름 지정 !cd /usr/local/lib/python3.8/dist-packages/pyspark/jars && wget https://s3.amazonaws.com/redshift-downloads/drivers/jdbc/1.2.20.1043/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar from pyspark.sql import SparkSession spark = SparkSession \ .builder \ .appName("Python Spark SQL #1") \ .getOrCreate() # Redshift와 연결 및 DataFrame으로 로딩하기 u..

실습 실습 2. 월별 채널별 매출과 방문자 정보 계산하기 테이블을 데이터프레임으로 로딩 데이터프레임별 테이블 이름 지정 !cd /usr/local/lib/python3.8/dist-packages/pyspark/jars && wget https://s3.amazonaws.com/redshift-downloads/drivers/jdbc/1.2.20.1043/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar from pyspark.sql import SparkSession spark = SparkSession \ .builder \ .appName("Python Spark SQL #1") \ .getOrCreate() # Redshift와 연결 및 DataFrame으로 로딩하기 url..