데이터 엔지니어 기술 블로그

About 세 가지 위치에서 스파크 설정을 할 수 있다. Spark properties: SparkConf 객체 사용, Java system properties 사용 val conf = new SparkConf() .setMaster("local[2]") .setAppName("CountingSheep") val sc = new SparkContext(conf) Environment variables: conf/spark-env.sh 환경 변수를 통하여 시스템별 설정 가능 # Options read when launching programs locally with # ./bin/run-example or ./bin/spark-submit # - HADOOP_CONF_DIR, to point Spark t..

Spark Streaming Kafka Option Example df = spark \ .readStream \ .format("kafka") \ .option("option name", "option value") \ .load() startingTimestamp 구독중인 주제의 모든 파티션에 대한 시작 타임스탬프를 지정한다. 타입: timestamp string e.g. "1000" 기본값: none (next preference is startingOffsetsByTimestamp) startingOffsetsByTimestamp 쿼리가 시작될 때 타임스탬프의 시작점 타입: json string """ {"topicA":{"0": 1000, "1": 1000}, "topicB": {"0": 200..

개요 카프카를 사용하는 경우 데이터를 가공해서 다시 카프카로 넣거나 다른 곳으로 보내는 등의 처리를 해줄 곳이 반드시 필요하다. Spark Streaming을 사용하면 문제를 쉽게 해결할 수 있다. 자체적으로 kafka에서 읽고 처리 후 kafka로 보내는 기능이 포함되어 있으며 Spark에서는 1.2버전부터 파이썬에서 Spark Streaming을 사용할 수 있게 되었다. Spark Streaming에는 DStreams라는 기능이 있고, 그 위에 DataFrame을 사용하여 더 쉽게 처리를 할 수 있는 Structed Streaming이 있다. 여기에서는 Structed Streaming을 사용하려고 한다. 방법 spark.readStream을 사용하여 카프카의 어떤 토픽에서 데이터를 가져올지 정한다...

개요 스파크를 실행하려고 할 때 권한 오류가 발생할 수 있다. 이 경우에는 AWS EMR의 Zeppelin에서 스크립트를 실행하는데에 오류가 발생했다. org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=WRITE, inode="/":hdfs:hdfsadmingroup:drwxr-xr-x 해결방법 권한을 검사하지 않게 하는 방법도 있지만, 간단한 해결 방법은 유저에게 권한을 주면 해결할 수 있다. dfs.permissions.superusergroup 을 보면 어떤 그룹이 superusergroup인지 확인할 수 있다. >> cat /etc/hadoop/conf/hdfs-site.xml ... dfs..

개요 에어플로우를 LocalExecutor를 사용해서 하나의 인스턴스에 실행시키고 있었다. 그러나 인스턴스에 문제가 생기면 Airflow도 동작하지 않을 수 있는 문제가 있고, 인프라가 커졌을 때 확장이 힘들어서 Worker를 분리해야했다. airflow에서 공식적으로 제공하는 docker-compose에서는 celery를 사용하며, 보통 airflow를 구축할 때 celery를 사용하는 것 같다. 이렇게 클러스터로 구축을 하게 되면 리소스가 필요할 때 worker의 갯수만 늘리면 되고, 리소스가 필요 없을 때 불필요하게 낭비하지 않고 worker의 갯수를 줄이면 되기 때문에 좋다. 구축할 때는 AWS ECS, AWS EFS 환경에서 구축했다. 구축 과정 Dask Cluster를 활용하여 구축 시도 가장..