Spark Streaming을 사용한 Apache Kafka 및 스트리밍 데이터 처리

안녕, 하브르! 오늘은 Spark Streaming을 사용하여 Apache Kafka 메시지 스트림을 처리하고 처리 결과를 AWS RDS 클라우드 데이터베이스에 쓰는 시스템을 구축하겠습니다.

특정 신용 기관이 모든 지점에서 "즉시" 들어오는 거래를 처리하는 작업을 우리에게 지정했다고 가정해 보겠습니다. 이는 재무부의 공개 통화 포지션, 거래 한도 또는 재무 결과 등을 신속하게 계산하기 위한 목적으로 수행될 수 있습니다.

마법과 마법 주문을 사용하지 않고 이 사례를 구현하는 방법 - 아래 내용을 읽어보세요! 가다!

(이미지 출처)

소개

물론 대량의 데이터를 실시간으로 처리하면 현대 시스템에서 사용할 수 있는 충분한 기회가 제공됩니다. 가장 널리 사용되는 조합 중 하나는 Apache Kafka와 Spark Streaming의 직렬 조합입니다. 여기서 Kafka는 수신 메시지 패킷의 스트림을 생성하고 Spark Streaming은 지정된 시간 간격으로 이러한 패킷을 처리합니다.

애플리케이션의 내결함성을 높이기 위해 체크포인트를 사용합니다. 이 메커니즘을 사용하면 Spark Streaming 엔진이 손실된 데이터를 복구해야 할 때 마지막 체크포인트로 돌아가서 거기에서 계산을 재개하기만 하면 됩니다.

개발된 시스템의 아키텍처

사용된 구성 요소:

• 아파치 카프카 분산 게시-구독 메시징 시스템입니다. 오프라인 및 온라인 메시지 소비 모두에 적합합니다. 데이터 손실을 방지하기 위해 Kafka 메시지는 디스크에 저장되고 클러스터 내에 복제됩니다. Kafka 시스템은 ZooKeeper 동기화 서비스 위에 구축되었습니다.
• 아파치 스파크 스트리밍 - 스트리밍 데이터 처리를 위한 Spark 구성 요소입니다. Spark Streaming 모듈은 데이터 스트림이 작은 데이터 패킷의 연속 시퀀스로 해석되는 마이크로 배치 아키텍처를 사용하여 구축되었습니다. Spark Streaming은 다양한 소스에서 데이터를 가져와 작은 패키지로 결합합니다. 새로운 패키지는 정기적으로 생성됩니다. 각 시간 간격이 시작될 때 새 패킷이 생성되고 해당 간격 동안 수신된 모든 데이터가 패킷에 포함됩니다. 간격이 끝나면 패킷 증가가 중지됩니다. 간격의 크기는 배치 간격이라는 매개변수에 의해 결정됩니다.
• 아파치 스파크 SQL - 관계형 처리와 Spark 함수형 프로그래밍을 결합합니다. 구조화된 데이터는 스키마, 즉 모든 레코드에 대한 단일 필드 집합이 있는 데이터를 의미합니다. Spark SQL은 다양한 구조화된 데이터 소스의 입력을 지원하며, 스키마 정보의 가용성 덕분에 필요한 레코드 필드만 효율적으로 검색할 수 있으며 DataFrame API도 제공합니다.
• AWS RDS 상대적으로 저렴한 클라우드 기반 관계형 데이터베이스, 설정, 운영 및 확장을 단순화하고 Amazon에서 직접 관리하는 웹 서비스입니다.

Kafka 서버 설치 및 실행

Kafka를 직접 사용하기 전에 Java가 있는지 확인해야 합니다. 왜냐하면... JVM은 작업에 사용됩니다.

sudo apt-get update 
sudo apt-get install default-jre
java -version

Kafka를 사용할 새 사용자를 만들어 보겠습니다.

sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo

다음으로 공식 Apache Kafka 웹사이트에서 배포판을 다운로드합니다.

wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"

다운로드한 아카이브의 압축을 푼다:

tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka

다음 단계는 선택사항입니다. 사실 기본 설정으로는 Apache Kafka의 모든 기능을 완전히 사용할 수 없습니다. 예를 들어 메시지를 게시할 수 있는 주제, 카테고리, 그룹을 삭제합니다. 이를 변경하려면 구성 파일을 편집해 보겠습니다.

vim ~/kafka/config/server.properties

파일 끝에 다음을 추가합니다.

delete.topic.enable = true

Kafka 서버를 시작하기 전에 ZooKeeper 서버를 시작해야 하며 Kafka 배포와 함께 제공되는 보조 스크립트를 사용합니다.

Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

ZooKeeper가 성공적으로 시작된 후 별도의 터미널에서 Kafka 서버를 시작합니다.

bin/kafka-server-start.sh config/server.properties

Transaction이라는 새 주제를 만들어 보겠습니다.

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

필요한 수의 파티션과 복제가 포함된 주제가 생성되었는지 확인해 보겠습니다.

bin/kafka-topics.sh --describe --zookeeper localhost:2181

새로 생성된 주제에 대해 생산자와 소비자를 테스트하는 순간을 놓치지 마세요. 메시지 보내기 및 받기를 테스트하는 방법에 대한 자세한 내용은 공식 문서에 기록되어 있습니다. 메시지 보내기. 이제 KafkaProducer API를 사용하여 Python으로 생산자를 작성하는 단계로 넘어갑니다.

프로듀서 글쓰기

생산자는 무작위 데이터(매초 100개의 메시지)를 생성합니다. 무작위 데이터란 세 가지 필드로 구성된 사전을 의미합니다.

• 지사 - 신용 기관의 판매 지점 이름
• 환율 — 거래 통화
• 금액 — 거래 금액. 금액은 은행이 통화를 구매하는 경우 양수, 판매인 경우 음수가 됩니다.

생산자의 코드는 다음과 같습니다.

from numpy.random import choice, randint

def get_random_value():
    new_dict = {}

    branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
    currency_list = ['RUB', 'USD', 'EUR', 'GBP']

    new_dict['branch'] = choice(branch_list)
    new_dict['currency'] = choice(currency_list)
    new_dict['amount'] = randint(-100, 100)

    return new_dict

다음으로, send 메소드를 사용하여 JSON 형식으로 필요한 주제에 대한 메시지를 서버에 보냅니다.

from kafka import KafkaProducer    

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                             value_serializer=lambda x:dumps(x).encode('utf-8'),
                             compression_type='gzip')
my_topic = 'transaction'
data = get_random_value()

try:
    future = producer.send(topic = my_topic, value = data)
    record_metadata = future.get(timeout=10)
    
    print('--> The message has been sent to a topic: 
            {}, partition: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> It seems an Error occurred: {}'.format(e))

finally:
    producer.flush()

스크립트를 실행하면 터미널에 다음 메시지가 표시됩니다.

이는 모든 것이 우리가 원하는 대로 작동한다는 것을 의미합니다. 생산자는 우리가 필요한 주제에 메시지를 생성하고 보냅니다.
다음 단계는 Spark를 설치하고 이 메시지 스트림을 처리하는 것입니다.

아파치 스파크 설치

아파치 스파크 범용 고성능 클러스터 컴퓨팅 플랫폼입니다.

Spark는 대화형 쿼리 및 스트림 처리를 포함하여 광범위한 계산 유형을 지원하면서 널리 사용되는 MapReduce 모델 구현보다 더 나은 성능을 발휘합니다. 속도는 대용량 데이터를 처리할 때 중요한 역할을 합니다. 몇 분, 몇 시간을 기다리지 않고도 대화형으로 작업할 수 있게 해주는 속도이기 때문입니다. Spark를 매우 빠르게 만드는 가장 큰 장점 중 하나는 메모리 내 계산을 수행하는 능력입니다.

이 프레임워크는 Scala로 작성되었으므로 먼저 설치해야 합니다.

sudo apt-get install scala

공식 웹사이트에서 Spark 배포판을 다운로드하세요.

wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"

아카이브의 압축을 풉니다.

sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark

bash 파일에 Spark 경로를 추가합니다.

vim ~/.bashrc

편집기를 통해 다음 줄을 추가합니다.

SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH

bashrc를 변경한 후 아래 명령을 실행하십시오.

source ~/.bashrc

AWS PostgreSQL 배포

남은 것은 스트림에서 처리된 정보를 업로드할 데이터베이스를 배포하는 것뿐입니다. 이를 위해 AWS RDS 서비스를 사용합니다.

AWS 콘솔 -> AWS RDS -> 데이터베이스 -> 데이터베이스 생성으로 이동합니다.

PostgreSQL을 선택하고 다음을 클릭합니다.

왜냐하면 이 예는 교육 목적으로만 사용되며 "최소"(무료 계층) 무료 서버를 사용합니다.

다음으로, 무료 계층 블록에 체크 표시를 하면 t2.micro 클래스의 인스턴스가 자동으로 제공됩니다. 비록 약하기는 하지만 무료이며 우리 작업에 매우 적합합니다.

다음은 매우 중요한 사항입니다. 데이터베이스 인스턴스 이름, 마스터 사용자 이름 및 비밀번호입니다. 인스턴스 이름을 myHabrTest, 마스터 사용자로 지정하겠습니다. 하브르, 비밀번호: 하브르12345 다음 버튼을 클릭하세요.

다음 페이지에는 외부에서 데이터베이스 서버에 대한 접근성(공용 접근성) 및 포트 가용성을 담당하는 매개변수가 있습니다.

포트 5432(PostgreSQL)를 통해 데이터베이스 서버에 대한 외부 액세스를 허용하는 VPC 보안 그룹에 대한 새 설정을 생성해 보겠습니다.
별도의 브라우저 창에서 AWS 콘솔로 이동하여 VPC 대시보드 -> 보안 그룹 -> 보안 그룹 생성 섹션을 살펴보겠습니다.

보안 그룹의 이름인 PostgreSQL을 설명으로 설정하고 이 그룹이 연결되어야 하는 VPC를 나타내고 생성 버튼을 클릭합니다.

아래 그림과 같이 새로 생성된 그룹에 대해 포트 5432에 대한 인바운드 규칙을 입력합니다. 포트를 수동으로 지정할 수는 없지만 유형 드롭다운 목록에서 PostgreSQL을 선택합니다.

엄밀히 말하면 ::/0 값은 전 세계에서 서버로 들어오는 트래픽의 가용성을 의미합니다. 이는 정식으로 완전히 사실은 아니지만 예제를 분석하기 위해 다음 접근 방식을 사용해 보겠습니다.

"고급 설정 구성"이 열려 있는 브라우저 페이지로 돌아가서 VPC 보안 그룹 섹션 -> 기존 VPC 보안 그룹 선택 -> PostgreSQL을 선택합니다.

다음으로 데이터베이스 옵션 -> 데이터베이스 이름 -> 이름 설정 - habrDB.

기본적으로 백업 비활성화(백업 보존 기간 - 0일), 모니터링 및 성능 개선 도우미를 제외한 나머지 매개 변수는 그대로 둘 수 있습니다. 버튼을 클릭하세요 데이터베이스 생성:

스레드 핸들러

마지막 단계는 XNUMX초마다 Kafka에서 들어오는 새로운 데이터를 처리하고 결과를 데이터베이스에 입력하는 Spark 작업의 개발입니다.

위에서 언급한 것처럼 체크포인트는 내결함성을 보장하기 위해 구성해야 하는 SparkStreaming의 핵심 메커니즘입니다. 체크포인트를 사용할 것이며, 절차가 실패할 경우 Spark Streaming 모듈은 마지막 체크포인트로 돌아가 계산을 재개하기만 하면 손실된 데이터를 복구할 수 있습니다.

체크포인트 정보가 저장될 내결함성, 신뢰성 있는 파일 시스템(예: HDFS, S3 등)에 디렉터리를 설정하여 체크포인트를 활성화할 수 있습니다. 이는 예를 들어 다음을 사용하여 수행됩니다.

streamingContext.checkpoint(checkpointDirectory)

이 예에서는 다음 접근 방식을 사용합니다. 즉, checkpointDirectory가 존재하는 경우 체크포인트 데이터에서 컨텍스트가 다시 생성됩니다. 디렉터리가 존재하지 않는 경우(즉, 처음으로 실행된 경우) functionToCreateContext가 호출되어 새 컨텍스트를 생성하고 DStreams를 구성합니다.

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

KafkaUtils 라이브러리의 createDirectStream 메서드를 사용하여 "트랜잭션" 주제에 연결하기 위한 DirectStream 개체를 만듭니다.

from pyspark.streaming.kafka import KafkaUtils
    
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)

broker_list = 'localhost:9092'
topic = 'transaction'

directKafkaStream = KafkaUtils.createDirectStream(ssc,
                                [topic],
                                {"metadata.broker.list": broker_list})

JSON 형식으로 들어오는 데이터를 구문 분석합니다.

rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
                                       currency=w['currency'],
                                       amount=w['amount']))
                                       
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")

Spark SQL을 사용하여 간단한 그룹화를 수행하고 결과를 콘솔에 표시합니다.

select 
    from_unixtime(unix_timestamp()) as curr_time,
    t.branch                        as branch_name,
    t.currency                      as currency_code,
    sum(amount)                     as batch_value
from treasury_stream t
group by
    t.branch,
    t.currency

쿼리 텍스트를 가져와 Spark SQL을 통해 실행합니다.

sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)

그런 다음 결과 집계 데이터를 AWS RDS의 테이블에 저장합니다. 집계 결과를 데이터베이스 테이블에 저장하기 위해 DataFrame 객체의 write 메소드를 사용합니다.

testResultDataFrame.write 
    .format("jdbc") 
    .mode("append") 
    .option("driver", 'org.postgresql.Driver') 
    .option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB") 
    .option("dbtable", "transaction_flow") 
    .option("user", "habr") 
    .option("password", "habr12345") 
    .save()

AWS RDS에 대한 연결 설정에 대한 몇 마디입니다. “AWS PostgreSQL 배포” 단계에서 이에 대한 사용자와 비밀번호를 생성했습니다. 연결 및 보안 섹션에 표시되는 데이터베이스 서버 URL로 Endpoint를 사용해야 합니다.

Spark와 Kafka를 올바르게 연결하려면 아티팩트를 사용하여 smark-submit을 통해 작업을 실행해야 합니다. 스파크 스트리밍-kafka-0-8_2.11. 또한 PostgreSQL 데이터베이스와 상호작용하기 위한 아티팩트도 사용할 것이며 --packages를 통해 이를 전송할 것입니다.

스크립트의 유연성을 위해 메시지 서버의 이름과 데이터를 수신하려는 주제도 입력 매개변수로 포함합니다.

이제 시스템 기능을 실행하고 확인할 차례입니다.

spark-submit 
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207 
spark_job.py localhost:9092 transaction

모든 것이 해결되었습니다! 아래 그림에서 볼 수 있듯이, StreamingContext 객체를 생성할 때 일괄 처리 간격을 2초로 설정했기 때문에 애플리케이션이 실행되는 동안 2초마다 새로운 집계 결과가 출력됩니다.

다음으로, 데이터베이스에 간단한 쿼리를 만들어 테이블에 레코드가 있는지 확인합니다. transaction_flow:

결론

이 기사에서는 Apache Kafka 및 PostgreSQL과 함께 Spark Streaming을 사용하여 정보 스트림 처리의 예를 살펴보았습니다. 다양한 소스의 데이터가 증가함에 따라 스트리밍 및 실시간 애플리케이션 제작에 있어 Spark Streaming의 실질적인 가치를 과대평가하기는 어렵습니다.

내 저장소에서 전체 소스 코드를 찾을 수 있습니다. GitHub의.

나는 이 기사에 대해 토론하게 되어 기쁘고, 여러분의 의견을 기대하며, 또한 관심을 갖고 있는 모든 독자들의 건설적인 비판을 바랍니다.

난 당신에게 성공을 기원합니다!

시. 처음에는 로컬 PostgreSQL 데이터베이스를 사용할 계획이었지만 AWS에 대한 애정을 고려하여 데이터베이스를 클라우드로 이동하기로 결정했습니다. 이 주제에 대한 다음 기사에서는 AWS Kinesis 및 AWS EMR을 사용하여 AWS에서 위에서 설명한 전체 시스템을 구현하는 방법을 보여 드리겠습니다. 뉴스를 팔로우하세요!

출처 : habr.com