집 >백엔드 개발 >파이썬 튜토리얼 >Spark Structured Streaming 및 Apache Kafka를 사용한 실시간 항공 교통 데이터 분석

Spark Structured Streaming 및 Apache Kafka를 사용한 실시간 항공 교통 데이터 분석

Mary-Kate Olsen원래의: 2024-10-29 21:55:02951검색

현재 우리는 매초 페타바이트의 데이터가 생성되는 세상에 살고 있습니다. 따라서 이 데이터를 실시간으로 분석하고 처리하는 것은 데이터가 생성되고 더 많은 데이터가 생성됨에 따라 보다 정확하게 비즈니스 통찰력을 생성하려는 기업에게 매우 중요합니다.

오늘은 Spark Structured Streaming과 Apache Kafka를 활용하여 가상의 항공 교통 데이터를 기반으로 실시간 데이터 분석을 개발해 보겠습니다. 이러한 기술이 무엇인지 모르신다면 제가 해당 기술을 더 자세히 소개한 기사와 이 기사 전체에서 다룰 다른 개념을 읽어 보시기 바랍니다. 그럼 잊지 말고 꼭 확인해보세요.

Análise de dados de tráfego aéreo em tempo real com Spark Structured Streaming e Apache Kafka

Spark 구조적 스트리밍 및 Apache Kafka를 사용한 실시간 데이터 처리에 대한 간략한 소개

게하지 앙크 ・'22년 9월 29일

#파이썬 #pyspark #데이터엔지니어링 #아파체카프카

제 GitHub에서 전체 프로젝트를 확인하실 수 있습니다.

건축학

글쎄요, 데이터 엔지니어인 당신이 SkyX라는 항공사에서 일하고 있다고 상상해 보세요. 그곳에서 항공 교통에 대한 데이터는 매초 생성됩니다.

해외에서 가장 많이 방문한 도시 순위 등 이러한 항공편의 실시간 데이터를 표시하는 대시보드를 개발해 달라는 요청을 받았습니다. 대부분의 사람들이 떠나는 도시; 그리고 전 세계에서 가장 많은 사람을 수송하는 항공기입니다.

각 항공편에서 생성되는 데이터는 다음과 같습니다.

aircraft_name: 항공기 이름입니다. SkyX에는 5대의 항공기만 이용 가능합니다.
출발지: 항공기가 출발하는 도시입니다. SkyX는 전 세계 5개 도시 간 항공편만 운항합니다.
도착지: 항공기 목적지 도시. 앞서 언급했듯이 SkyX는 전 세계 5개 도시 간 항공편만 운항합니다.
승객: 항공기가 수송하는 승객 수입니다. 모든 SkyX 항공기는 각 항공편당 50~100명을 수송합니다.

다음은 우리 프로젝트의 기본 아키텍처입니다.

생산자: 항공기 항공 교통 데이터를 생성하여 Apache Kafka 주제로 전송하는 역할을 담당합니다.
소비자: Apache Kafka 주제에 실시간으로 도착하는 데이터만 관찰합니다.
데이터 분석: Apache Kafka 주제에 도달하는 데이터를 실시간으로 처리하고 분석하는 3개의 대시보드입니다. 가장 많은 관광객이 방문하는 도시를 분석합니다. 대부분의 사람들이 다른 도시를 방문하기 위해 떠나는 도시 분석; 전 세계 도시 간 가장 많은 사람을 수송하는 SkyX 항공기 분석

개발 환경 준비

이 튜토리얼에서는 컴퓨터에 PySpark가 이미 설치되어 있다고 가정합니다. 아직 확인하지 않았다면 설명서 자체의 단계를 확인하세요.

Apache Kafka는 Docker??를 통한 컨테이너화로 사용하겠습니다.

그리고 마지막으로 가상환경을 통해 Python을 사용해보겠습니다.

Docker를 통한 컨테이너화를 통한 Apache Kafka

더 이상 고민하지 말고 skyx라는 폴더를 만들고 그 안에 docker-compose.yml 파일을 추가하세요.

$ mkdir skyx
$ cd skyx
$ touch docker-compose.yml

이제 docker-compose 파일에 다음 콘텐츠를 추가하세요.

version: '3.9'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

완료! 이제 Kafka 서버를 업로드할 수 있습니다. 이렇게 하려면 터미널에 다음 명령을 입력하세요.

$ docker compose up -d
$ docker compose ps

NAME                                COMMAND                  SERVICE             STATUS              PORTS
skyx-kafka-1       "/etc/confluent/dock…"   kafka               running             9092/tcp, 0.0.0.0:29092->29092/tcp
skyx-zookeeper-1   "/etc/confluent/dock…"   zookeeper           running             2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp

참고: 이 튜토리얼에서는 Docker Compose 버전 2.0을 사용합니다. 이것이 docker와 compose 사이에 "-"가 없는 이유입니다. ☺.

이제 생산자가 실시간으로 보낸 데이터를 저장할 Kafka 내 주제를 생성해야 합니다. 이를 위해 컨테이너 내부의 Kafka에 액세스해 보겠습니다.

$ docker compose exec kafka bash

마지막으로 airtraffic이라는 주제를 만듭니다.

$ kafka-topics --create --topic airtraffic --bootstrap-server localhost:29092

항공교통 주제를 만들었습니다.

가상 환경 생성

실시간 항공 교통 데이터를 Kafka 주제로 전송하는 프로듀서, 즉 애플리케이션을 개발하려면 kafka-python 라이브러리를 사용해야 합니다. kafka-python은 Apache Kafka와 통합되는 생산자와 소비자를 개발할 수 있는 커뮤니티 개발 라이브러리입니다.

먼저 requirements.txt라는 파일을 만들고 그 안에 다음 종속성을 추가해 보겠습니다.

카프카파이썬

두 번째로 가상 환경을 생성하고 요구사항.txt 파일에 종속성을 설치합니다.

$ mkdir skyx
$ cd skyx
$ touch docker-compose.yml

완료! 이제 우리 환경은 개발 준비가 되었나요?.

생산자 개발

이제 프로듀서를 만들어 보겠습니다. 앞서 언급했듯이 제작자는 새로 생성된 Kafka 주제에 항공 교통 데이터를 전송하는 책임을 맡게 됩니다.

건축에서도 말했듯이 SkyX는 전 세계 5개 도시 사이만 비행하며, 이용 가능한 항공기는 5대뿐인가요? 각 항공기에는 50~100명의 승객이 탑승한다는 점은 언급할 가치가 있습니다.

데이터는 무작위로 생성되어 1~6초 사이의 시간 간격으로 json 형식으로 주제에 전송됩니다.

가자! src라는 하위 디렉터리와 kafka라는 다른 하위 디렉터리를 만듭니다. kafka 디렉터리 내에 airtraffic_producer.py라는 파일을 만들고 그 안에 다음 코드를 추가합니다.

version: '3.9'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

완료! 우리는 생산자를 개발합니다. 실행하고 잠시 동안 실행해 보세요.

$ python airtraffic_producer.py

소비자 개발

이제 소비자를 개발해 보겠습니다. 이것은 매우 간단한 응용 프로그램이 될 것입니다. kafka 주제에 도착하는 데이터를 실시간으로 터미널에 표시합니다.

kafka 디렉터리 내에서 airtraffic_consumer.py라는 파일을 만들고 그 안에 다음 코드를 추가합니다.

$ docker compose up -d
$ docker compose ps

보세요, 아주 간단하다고 말씀드렸죠. 실행하여 제작자가 주제에 데이터를 보내면서 실시간으로 표시되는 데이터를 지켜보세요.

$ python airtraffic_consumer.py

데이터 분석: 관광객이 가장 많은 도시

이제 데이터 분석부터 시작하겠습니다. 이 시점에서 우리는 가장 많은 관광객을 유치하는 도시의 순위를 실시간으로 표시하는 대시보드, 애플리케이션을 개발할 것입니다. 즉, to 열을 기준으로 데이터를 그룹화하고 승객 열을 기준으로 합계를 계산합니다. 아주 간단해요!

이를 수행하려면 src 디렉터리 내에 dashboards라는 하위 디렉터리를 만들고 tourists_analytic.py라는 파일을 만듭니다. 그런 다음 그 안에 다음 코드를 추가하세요.

$ mkdir skyx
$ cd skyx
$ touch docker-compose.yml

이제 Spark-Submit을 통해 파일을 실행할 수 있습니다. 하지만 진정하세요! PySpark를 Kafka와 통합할 때 스파크 제출을 다르게 실행해야 합니다. --packages.

매개변수를 통해 Apache Kafka 패키지와 Apache Spark의 현재 버전을 알려야 합니다.

Apache Spark를 Apache Kafka와 처음 통합하는 경우 Spark-submit을 실행하는 데 시간이 걸릴 수 있습니다. 필요한 패키지를 다운로드 받아야 하기 때문입니다.

데이터 분석을 실시간으로 볼 수 있도록 생산자가 여전히 실행 중인지 확인하세요. 대시보드 디렉터리 내에서 다음 명령을 실행합니다.

$ Spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 tourist_analytic.py

version: '3.9'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

데이터 분석: 대부분의 사람들이 떠나는 도시

이번 분석은 이전 분석과 매우 유사합니다. 하지만, 관광객이 가장 많이 들어오는 도시를 실시간으로 분석하는 대신, 사람들이 가장 많이 떠나는 도시를 분석하겠습니다. 이렇게 하려면 leavers_analytic.py라는 파일을 만들고 그 안에 다음 코드를 추가하세요.

$ docker compose up -d
$ docker compose ps

데이터 분석을 실시간으로 볼 수 있도록 생산자가 여전히 실행 중인지 확인하세요. 대시보드 디렉터리 내에서 다음 명령을 실행합니다.

$ Spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 Leavers_analytic.py

NAME                                COMMAND                  SERVICE             STATUS              PORTS
skyx-kafka-1       "/etc/confluent/dock…"   kafka               running             9092/tcp, 0.0.0.0:29092->29092/tcp
skyx-zookeeper-1   "/etc/confluent/dock…"   zookeeper           running             2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp

데이터 분석: 가장 많은 승객을 태운 항공기

이번 분석은 이전 분석보다 훨씬 간단합니다. 전 세계 도시 간 가장 많은 승객을 수송하는 항공기를 실시간으로 분석해 보겠습니다. aircrafts_analytic.py라는 파일을 만들고 그 안에 다음 코드를 추가하세요.

$ python -m venv venv
$ venv\scripts\activate
$ pip install -r requirements.txt

데이터 분석을 실시간으로 볼 수 있도록 생산자가 여전히 실행 중인지 확인하세요. 대시보드 디렉터리 내에서 다음 명령을 실행합니다.

$ Spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 Aircrafts_analytic.py

$ mkdir skyx
$ cd skyx
$ touch docker-compose.yml

최종 고려사항

여기서 마치겠습니다! 이 기사에서는 Spark Structured Streaming 및 Apache Kafka를 사용하여 가상의 항공 교통 데이터를 기반으로 실시간 데이터 분석을 개발합니다.

이를 위해 우리는 이 데이터를 Kafka 주제에 실시간으로 전송하는 생산자를 개발한 다음, 이 데이터를 실시간으로 분석하기 위해 3개의 대시보드를 개발했습니다.

좋아하셨으면 좋겠습니다. 다음에 또 만나요?.

위 내용은 Spark Structured Streaming 및 Apache Kafka를 사용한 실시간 항공 교통 데이터 분석의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

Python bash sql json bootstrap kafka tr github docker spark apache

성명：

이전 기사：두 Python 사전에서 일치하는(키, 값) 쌍을 어떻게 효율적으로 찾을 수 있습니까?다음 기사：두 Python 사전에서 일치하는(키, 값) 쌍을 어떻게 효율적으로 찾을 수 있습니까?