예니의 코딩일기

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

강의 AWS Spark AWS에서 Spark를 실행하기 위해서는 EMR(Elastic MapReduce)를 이용하는 것이 일반적입니다. "EMR"은 Hadoop의 YARN과 유사한 기능을 제공하는 AWS의 서비스 입니다. worker node : EC2 서버 HDFS : S3 실행 과정 AWS EMR 클러스터 생성 옵션 : Spark 실행 m5.xlarge 노드 3개 선택 기본 파일 시스텀 선택 : S3 마스터 노드 포트 번호 22 지정 마스터 노드를 드라이버 노드로 사용 : SSH 이용하여 로그인 PySpark 잡 코드 실행 ssh -i 프라이빗키.pen hadoop@마스터노드호스트이름 spark-submit --master yarn stackoverflow.py # stackoverflow.py fr..

강의 내부 연산 Transformations Narrow Dependencies 독립적인 Partition level 작업 병렬처리 가능 select, where, filter, map, count Wide Depedencies Shuffling이 필요한 작업 groupby, reduceby, partitionby, repartition, coalece Actions Read, Write, Show, Collect Job 실행 Lazy Execution 내부 동작 Action → Job → Stages → Tasks Action : 하나의 Job 생성 및 코드 실행 Jobs : 다수의 Stage로 구성 후 Shuffling 발생시 추가 생성 Stages : DAG 형태의 Task 병렬 실행 가능 Task..

강의 Spark 파일 포맷 Unstructured : Text 눈으로 확인 가능 Semi-structured : JSON, WML, CSV 눈으로 확인 가능 Structured : PARQUET, AVRO, ORC, SequenceFile 압축된 정보 스키마 제공 # 파일 다운로드 !wget https://pyspark-test-sj.s3.us-west-2.amazonaws.com/appl_stock.csv from pyspark.sql import * from pyspark.sql.functions import * if __name__ == "__main__": spark = SparkSession \ .builder \ .appName("Spark Writing Demo") \ .master("loc..

강의 Unit Test "Unit Test"는 코드 상의 특정 기능을 테스트하기 위해 작성된 코드 입니다. 정해진 입력에 대한 예상된 출력 확인 형태 CI/CD를 사용할 때 전체 코드의 테스트 커버리지 중요 언어별 정해진 테스트 프레임워크 사용 Java : JUnit .NET : NUnit Python : unittest 실습 # 파일 다운로드 !wget https://s3-geospatial.s3-us-west-2.amazonaws.com/name_gender.csv from pyspark.sql import SparkSession spark = SparkSession \ .builder \ .appName("Python Spark Unit Test") \ .getOrCreate() df = spark..