일 | 월 | 화 | 수 | 목 | 금 | 토 |
---|---|---|---|---|---|---|
1 | 2 | 3 | 4 | |||
5 | 6 | 7 | 8 | 9 | 10 | 11 |
12 | 13 | 14 | 15 | 16 | 17 | 18 |
19 | 20 | 21 | 22 | 23 | 24 | 25 |
26 | 27 | 28 | 29 | 30 | 31 |
- 에어플로우
- Schema Registry
- 레드시프트
- Zookeeper
- spark
- Data Warehouse
- 컬럼 기반
- AWS
- Data engineering
- delta lake
- kafka
- 카프카 구축
- spark streaming
- 델타레이크
- Parquet
- Data Engineer
- docker
- 스파크
- MySQL
- 대용량 처리
- 스파크 스트리밍
- 데이터 웨어하우스
- airflow
- 데이터
- 데이터 엔지니어링
- s3
- Redshift
- kafka rest api
- 데이터 엔지니어
- 카프카
- Today
- Total
목록분류 전체보기 (62)
데이터 엔지니어 기술 블로그
About 트리노는 빅데이터를 쿼리하기 위한 분산 SQL 쿼리 엔진이다. 기존에는 HDFS를 쿼리하도록 설계되었지만 지금은 그것에만 국한되지 않는다. Trino라는 이름은 Presto라는 이름을 리브랜딩 한 것으로 같은 쿼리 엔진이다. 트리노는 범용 데이터베이스가 아니기 때문에 MySQL과 같이 OLTP(Online Transaction Processing)가 아닌 OLAP(Online Analytics Processing)로 설계되었다. Concepts Server types 트리노는 두 가지의 서버 타입이 있다. 하나는 Coordinator로 다른 하나는 Worker이다. 이름과 같이 Coordinator는 구문 분석, 쿼리 계획, 작업자 노드 관리와 같은 일을 하고 Worker는 실제 작업을 실행하..
개발 동기 우리는 어떤 하둡 에코시스템에도 사용할 수 있고 압축과 컬럼 기반 데이터 표현의 이점을 만들기 위해 Parquet를 개발했습니다. Parquet는 처음부터 중첩된(nested) 데이터 구조를 위해 개발되었으며, Dremel의 논문에 작성된 record shredding and assembly algorithm을 사용합니다. 우리는 이 접근 방식이 중첩된 name space를 단순하게 붙이는 것보다 좋다고 믿고 있습니다. https://github.com/julienledem/redelm/wiki/The-striping-and-assembly-algorithms-from-the-Dremel-paper Parquet는 효율적인 압축과 인코딩 체계를 지원하도록 개발되었습니다. 여러 프로젝트에서 데이..
개요 아파치 쓰리프트는 페이스북에서 서로 다른 언어간의 통신을 위하여 개발되었다. 원격 프로시저 호출(Remote Procedure Call)로 언어에 상관 없이 서로 통신할 수 있도록 도와준다. 예를 들면 PHP에서 작성한 기능을 파이썬과 Go언어에서 자유롭게 호출해서 사용할 수 있다. 단순히 하나의 함수 호출이 아니라, REST API 서버처럼 자유롭게 개발을 할 수 있다. thrift와 호환되는 언어들은 모두 비슷하지만 각자 다른 인터페이스 정의 규칙을 가지고 있다. 아파치 쓰리프트에서는 .thrift 파일에 변수의 타입과 이름 그리고 함수의 매개변수, 반환값과 예외 등의 정의를 하고, thrift를 사용하여 옵션으로 변환하려는 언어를 적어주면 .thrift 파일에 입력된 정의로 각 언어에 맞게 코..
에러 메세지 java.lang.AssertionError: assertion failed: Concurrent update to the log. Multiple streaming jobs detected for 53 at scala.Predef$.assert(Predef.scala:223) 원인 로그에 대한 동시 업데이트, 여러 스트리밍 작업이 발견되었습니다. 스파크 스트리밍에서 동일한 체크포인트를 사용하는 두 개의 싱크 작업(writeStream)이 동시에 실행되면 발생하는 문제이다. checkpointLocation을 다른 위치로 사용하면 해결할 수 있다. Zeppelin에서 코드를 실행하고 같은 스트리밍 스크립트를 사용할 때 오류가 발생할 수 있다. 기존 같은 location을 사용한다. 동시에 같..
전체 에러 메세지 StreamingQueryException: Cannot find earliest offsets of Set(topic-name-0). Some data may have been missed. Some data may have been lost because they are not available in Kafka any more; either the data was aged out by Kafka or the topic may have been deleted before all the data in the topic was processed. If you don't want your streaming query to fail on such cases, set the source opt..
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를 활용하여 구축 시도 가장..