Azure Databricks를 사용하는 스트림 처리

Azure Cosmos DB

Azure Databricks

Azure Event Hubs

Azure Log Analytics

Azure Monitor

이 참조 아키텍처는 엔드투엔드 스트림 처리 파이프라인을 보여줍니다. 이 파이프라인의 네 단계는 수집, 처리, 저장 및 분석 및 보고합니다. 이 참조 아키텍처의 경우 파이프라인은 실시간으로 두 원본에서 데이터를 수집하고, 각 스트림의 관련 레코드에 대해 조인을 수행하고, 결과를 보강하고, 평균을 계산합니다. 그런 다음 추가 분석을 위해 결과가 저장됩니다.

GitHub 로고 이 아키텍처에 대한 참조 구현은 GitHub에서 사용할 수 있습니다.

아키텍처

Azure Databricks를 사용한 스트림 처리를 위한 참조 아키텍처를 보여 주는

이 아키텍처의 Visio 파일 다운로드합니다.

워크플로

다음 데이터 흐름은 이전 다이어그램에 해당합니다.

이 아키텍처에는 실시간으로 데이터 스트림을 생성하는 두 개의 데이터 원본이 있습니다. 첫 번째 스트림에는 승차 정보가 포함되고 두 번째 스트림에는 요금 정보가 포함됩니다. 참조 아키텍처에는 정적 파일 집합에서 읽고 Azure Event Hubs에 데이터를 푸시하는 시뮬레이션된 데이터 생성기가 포함되어 있습니다. 실제 애플리케이션의 데이터 원본은 택시에 설치된 디바이스입니다.
Event Hubs 이벤트 수집 서비스입니다. 이 아키텍처는 각 데이터 원본에 대해 하나씩 두 개의 이벤트 허브 인스턴스를 사용합니다. 각 데이터 원본은 연결된 이벤트 허브에 데이터 스트림을 보냅니다.
Azure Databricks Microsoft Azure 클라우드 서비스 플랫폼에 최적화된 Apache Spark 기반 분석 플랫폼입니다. Azure Databricks는 택시 승차 및 요금 데이터의 상관 관계를 지정하고 Azure Databricks 파일 시스템에 저장된 이웃 데이터와 상관 관계가 있는 데이터를 보강하는 데 사용됩니다.
Azure Cosmos DB 완전히 관리되는 다중 모델 데이터베이스 서비스입니다. Azure Databricks 작업의 출력은 Azure Cosmos DB for Apache Cassandra에 기록되는 일련의 레코드입니다. Azure Cosmos DB for Apache Cassandra는 시계열 데이터 모델링을 지원하기 때문에 사용됩니다.
- Azure Cosmos DB용 Azure Synapse Link트랜잭션 워크로드에 대한 성능 또는 비용 영향 없이 Azure Cosmos DB의 운영 데이터에 대해 거의 실시간으로 분석을 실행할 수 있습니다. 서버리스 SQL 풀 Spark 풀 사용하여 이러한 결과를 얻을 수 있습니다. 이러한 분석 엔진은 Azure Synapse Analytics 작업 영역에서 사용할 수 있습니다.
- Microsoft Fabric NoSQL용 Azure Cosmos DB 미러링을 사용하면 Azure Cosmos DB 데이터를 Microsoft Fabric의 나머지 데이터와 통합할 수 있습니다.
Log Analytics 다양한 원본에서 로그 데이터를 쿼리하고 분석할 수 있는 Azure Monitor 내의 도구입니다. Azure Monitor 수집하는 애플리케이션 로그 데이터는 Log Analytics 작업 영역저장됩니다. Log Analytics 쿼리를 사용하여 메트릭을 분석 및 시각화하고 로그 메시지를 검사하여 애플리케이션 내에서 문제를 식별할 수 있습니다.

시나리오 정보

택시 회사는 각 택시 여정에 대한 데이터를 수집합니다. 이 시나리오에서는 두 개의 개별 디바이스가 데이터를 전송한다고 가정합니다. 택시에는 기간, 거리, 승차 및 하차 위치를 포함하여 각 승차에 대한 정보를 보내는 미터가 있습니다. 별도 디바이스는 고객의 지불을 수락하고 요금에 대한 데이터를 보냅니다. 택시 회사는 라이더십 추세를 파악하기 위해 각 지역의 주행 마일당 평균 팁을 실시간으로 계산하려고 합니다.

데이터 수집

데이터 원본을 시뮬레이션하기 위해 이 참조 아키텍처는 뉴욕시 택시 데이터 세트¹사용합니다. 이 데이터 세트에는 2010년부터 2013년까지 뉴욕시의 택시 여행에 대한 데이터가 포함되어 있습니다. 여기에는 승차 및 요금 데이터 레코드가 모두 포함됩니다. 승차 데이터에는 여정 기간, 여정 거리, 승차 및 하차 위치가 포함됩니다. 요금 데이터에는 요금, 세금 및 팁 금액이 포함됩니다. 두 레코드 유형의 필드에는 medallion 번호, 해킹 라이선스 및 공급업체 ID가 포함됩니다. 이 세 필드의 조합은 택시와 드라이버를 고유하게 식별합니다. 데이터가 CSV 형식으로 저장됩니다.

[1] Donovan, Brian; Work, Dan (2016): 뉴욕시 택시 여정 데이터(2010-2013) University of Illinois at Urbana-Champaign https://doi.org/10.13012/J8PN93H8

데이터 생성기는 레코드를 읽고 Event Hubs로 보내는 .NET Core 애플리케이션입니다. 생성기는 승객 데이터를 JSON 형식으로 보내고, 요금 데이터를 CSV 형식으로 전송합니다.

Event Hubs는 파티션을 사용하여 데이터를 분할합니다. 파티션을 사용하면 소비자가 각 파티션을 병렬로 읽을 수 있습니다. Event Hubs로 데이터를 보낼 때 파티션 키를 직접 지정할 수 있습니다. 그렇지 않으면 레코드는 라운드 로빈 방식으로 파티션에 할당됩니다.

이 시나리오에서는 승차 데이터 및 요금 데이터에 특정 택시에 대해 동일한 파티션 ID를 할당해야 합니다. 이 할당을 사용하면 Databricks가 두 스트림의 상관 관계를 지정할 때 병렬 처리 수준을 적용할 수 있습니다. 예를 들어, 파티션 n 승차 데이터의 레코드는 요금 데이터의 n 파티션의 레코드와 일치합니다.

이 아키텍처의 Visio 파일을 다운로드합니다.

데이터 생성기에서 두 레코드 형식에 대한 공통 데이터 모델에는 PartitionKey,Medallion, 및 HackLicense의 연결인 VendorId속성이 있습니다.

public abstract class TaxiData
{
    public TaxiData()
    {
    }

    [JsonProperty]
    public long Medallion { get; set; }

    [JsonProperty]
    public long HackLicense { get; set; }

    [JsonProperty]
    public string VendorId { get; set; }

    [JsonProperty]
    public DateTimeOffset PickupTime { get; set; }

    [JsonIgnore]
    public string PartitionKey
    {
        get => $"{Medallion}_{HackLicense}_{VendorId}";
    }

이 속성은 Event Hubs에 데이터를 보낼 때 명시적 파티션 키를 제공하는 데 사용됩니다.

using (var client = pool.GetObject())
{
    return client.Value.SendAsync(new EventData(Encoding.UTF8.GetBytes(
        t.GetData(dataFormat))), t.PartitionKey);
}

Event Hubs

Event Hubs의 처리량 용량은 처리량 단위로 제어됩니다. 자동 확장사용하도록 설정하여 이벤트 허브를 자동 크기 조정할 수 있습니다. 이 기능은 트래픽에 따라 처리량 단위의 크기를 구성된 최대값까지 자동으로 조정합니다.

스트림 처리

Azure Databricks에서 작업은 데이터 처리를 수행합니다. 작업이 클러스터에 할당된 다음, 클러스터에서 실행됩니다. 작업은 Java로 작성된 사용자 지정 코드 또는 Spark Notebook수 있습니다.

이 참조 아키텍처에서 작업은 Java 및 Scala로 작성된 클래스가 있는 Java 보관 파일입니다. Databricks 작업에 대한 Java 보관 파일을 지정하면 Databricks 클러스터는 작업에 대한 클래스를 지정합니다. 여기서 클래스의 main방법com.microsoft.pnp.TaxiCabReader에는 데이터 처리 논리가 포함됩니다.

두 이벤트 허브 인스턴스에서 스트림 읽기

데이터 처리 논리는 Spark 구조적 스트리밍을 사용하여 두 Azure 이벤트 허브 인스턴스에서 스트림을 읽습니다.

// Create a token credential using Managed Identity
val credential = new DefaultAzureCredentialBuilder().build()

val rideEventHubOptions = EventHubsConf(rideEventHubEntraIdAuthConnectionString)
  .setTokenProvider(EventHubsUtils.buildTokenProvider(..., credential))
  .setConsumerGroup(conf.taxiRideConsumerGroup())
  .setStartingPosition(EventPosition.fromStartOfStream)
val rideEvents = spark.readStream
  .format("eventhubs")
  .options(rideEventHubOptions.toMap)
  .load

val fareEventHubOptions = EventHubsConf(fareEventHubEntraIdAuthConnectionString)
  .setTokenProvider(EventHubsUtils.buildTokenProvider(..., credential))
  .setConsumerGroup(conf.taxiFareConsumerGroup())
  .setStartingPosition(EventPosition.fromStartOfStream)
val fareEvents = spark.readStream
  .format("eventhubs")
  .options(fareEventHubOptions.toMap)
  .load

이웃 정보를 사용하여 데이터 보강

승차 데이터에는 승차 및 하차 위치의 위도 및 경도 좌표가 포함됩니다. 이러한 좌표는 유용하지만 분석에는 쉽게 사용되지 않습니다. 따라서 이 데이터는 셰이프 파일읽은 이웃 데이터로 보강됩니다.

셰이프 파일 형식은 이진 형식이며 쉽게 구문 분석되지 않습니다. 그러나 GeoTools 라이브러리는 셰이프 파일 형식을 사용하는 지리 공간적 데이터에 대한 도구를 제공합니다. 이 라이브러리는 com.microsoft.pnp.GeoFinder 클래스에서 픽업 및 하차 위치의 좌표를 기반으로 이웃 이름을 결정하는 데 사용됩니다.

val neighborhoodFinder = (lon: Double, lat: Double) => {
      NeighborhoodFinder.getNeighborhood(lon, lat).get()
    }

승차 및 요금 데이터 조인

먼저 승객 및 요금 데이터가 변환됩니다.

val rides = transformedRides
  .filter(r => {
    if (r.isNullAt(r.fieldIndex("errorMessage"))) {
      true
    }
    else {
      malformedRides.add(1)
      false
    }
  })
  .select(
    $"ride.*",
    to_neighborhood($"ride.pickupLon", $"ride.pickupLat")
      .as("pickupNeighborhood"),
    to_neighborhood($"ride.dropoffLon", $"ride.dropoffLat")
      .as("dropoffNeighborhood")
  )
  .withWatermark("pickupTime", conf.taxiRideWatermarkInterval())

val fares = transformedFares
  .filter(r => {
    if (r.isNullAt(r.fieldIndex("errorMessage"))) {
      true
    }
    else {
      malformedFares.add(1)
      false
    }
  })
  .select(
    $"fare.*",
    $"pickupTime"
  )
  .withWatermark("pickupTime", conf.taxiFareWatermarkInterval())

그러면 승차 데이터가 요금 데이터와 조인됩니다.

val mergedTaxiTrip = rides.join(fares, Seq("medallion", "hackLicense", "vendorId", "pickupTime"))

데이터를 처리하고 Azure Cosmos DB에 삽입

각 지역의 평균 요금 금액은 특정 시간 간격으로 계산됩니다.

val maxAvgFarePerNeighborhood = mergedTaxiTrip.selectExpr("medallion", "hackLicense", "vendorId", "pickupTime", "rateCode", "storeAndForwardFlag", "dropoffTime", "passengerCount", "tripTimeInSeconds", "tripDistanceInMiles", "pickupLon", "pickupLat", "dropoffLon", "dropoffLat", "paymentType", "fareAmount", "surcharge", "mtaTax", "tipAmount", "tollsAmount", "totalAmount", "pickupNeighborhood", "dropoffNeighborhood")
      .groupBy(window($"pickupTime", conf.windowInterval()), $"pickupNeighborhood")
      .agg(
        count("*").as("rideCount"),
        sum($"fareAmount").as("totalFareAmount"),
        sum($"tipAmount").as("totalTipAmount"),
        (sum($"fareAmount")/count("*")).as("averageFareAmount"),
        (sum($"tipAmount")/count("*")).as("averageTipAmount")
      )
      .select($"window.start", $"window.end", $"pickupNeighborhood", $"rideCount", $"totalFareAmount", $"totalTipAmount", $"averageFareAmount", $"averageTipAmount")

그러면 평균 요금 금액이 Azure Cosmos DB에 삽입됩니다.

maxAvgFarePerNeighborhood
      .writeStream
      .queryName("maxAvgFarePerNeighborhood_cassandra_insert")
      .outputMode(OutputMode.Append())
      .foreach(new CassandraSinkForeach(connector))
      .start()
      .awaitTermination()

고려 사항

이러한 고려 사항은 워크로드의 품질을 향상시키는 데 사용할 수 있는 일련의 기본 원칙인 Azure Well-Architected Framework의 핵심 요소를 구현합니다. 자세한 내용은 Microsoft Azure Well-Architected Framework를 참조하세요.

보안

보안은 의도적인 공격 및 중요한 데이터 및 시스템의 오용에 대한 보증을 제공합니다. 자세한 내용은 보안대한 디자인 검토 검사 목록을 참조하세요.

Azure Databricks 작업 영역에 대한 액세스는 관리자 콘솔사용하여 제어됩니다. 관리자 콘솔에는 사용자를 추가하고, 사용자 권한을 관리하고, Single Sign-On을 설정하는 기능이 포함되어 있습니다. 관리자 콘솔을 통해 작업 영역, 클러스터, 작업 및 테이블에 대한 액세스 제어를 설정할 수도 있습니다.

비밀 관리

Azure Databricks에는 자격 증명을 저장하고 Notebook 및 작업에서 참조하는 데 사용되는 비밀 저장소 포함되어 있습니다. Azure Databricks 비밀 저장소 내에서 파티션 비밀의 범위를 지정합니다.

databricks secrets create-scope --scope "azure-databricks-job"

비밀은 범위 수준에서 추가됩니다.

databricks secrets put --scope "azure-databricks-job" --key "taxi-ride"

참고 항목

네이티브 Azure Databricks 범위 대신 Azure Key Vault 지원 범위 사용합니다.

코드에서 비밀은 Azure Databricks 비밀 유틸리티를 통해 액세스됩니다.

비용 최적화

비용 최적화는 불필요한 비용을 줄이고 운영 효율성을 개선하는 방법에 중점을 둡니다. 자세한 내용은 비용 최적화대한 디자인 검토 검사 목록을 참조하세요.

Azure 가격 계산기를 사용하여 비용을 예측합니다. 이 참조 아키텍처에 사용되는 다음 서비스를 고려합니다.

Event Hubs 비용 고려 사항

이 참조 아키텍처는 표준 계층에 Event Hubs를 배포합니다. 가격 책정 모델은 처리량 단위, 수신 이벤트, 캡처 이벤트를 기반으로 합니다. 수신 이벤트는 64KB 이하의 데이터 단위입니다. 더 큰 메시지는 64KB의 배수로 청구됩니다. Azure Portal 또는 Event Hubs 관리 API를 통해 처리량 단위를 지정합니다.

보존 기간이 더 필요한 경우 전용 계층을 고려하세요. 이 계층은 엄격한 요구 사항이 있는 단일 테넌트 배포를 제공합니다. 이 제품은 용량 단위를 기반으로 하며 처리량 단위에 종속되지 않는 클러스터를 빌드합니다. 또한 표준 계층은 수신 이벤트 및 처리량 단위에 따라 요금이 청구됩니다.

자세한 내용은 Event Hubs 가격 책정참조하세요.

Azure Databricks 비용 고려 사항

Azure Databricks는 표준 계층과 프리미엄 계층을 제공하며, 둘 다 세 개의 워크로드를 지원합니다. 이 참조 아키텍처는 프리미엄 계층에서 Azure Databricks 작업 영역을 배포합니다.

데이터 엔지니어링 워크로드는 작업 클러스터에서 실행되어야 합니다. 데이터 엔지니어는 클러스터를 사용하여 작업을 빌드하고 수행합니다. 데이터 분석 워크로드는 다목적 클러스터에서 실행되어야 하며 데이터 과학자가 데이터와 인사이트를 대화형으로 탐색, 시각화, 조작 및 공유하기 위한 것입니다.

Azure Databricks는 여러 가격 책정 모델을 제공합니다.

종량제 요금제

클러스터에서 프로비전된 VM(가상 머신) 및 선택한 VM 인스턴스에 따라 Azure DTU(Databricks 단위)에 대한 요금이 청구됩니다. DBU는 초당 사용량으로 청구되는 처리 기능 단위입니다. DBU 사용량은 Azure Databricks에서 실행되는 인스턴스의 크기 및 유형에 따라 달라집니다. 가격 책정은 선택한 워크로드 및 계층에 따라 달라집니다.
사전 구매 플랜

종량제 모델과 비교할 때 해당 기간 동안 총 소유 비용을 줄이기 위해 1~3년 동안 Azure Databricks 커밋 단위로 DPU를 커밋합니다.

자세한 내용은 Azure Databricks 가격 책정 참조하세요.

Azure Cosmos DB 비용 고려 사항

이 아키텍처에서 Azure Databricks 작업은 일련의 레코드를 Azure Cosmos DB에 씁니다. 예약한 용량에 대한 요금이 청구되며, 초당 요청 단위(RU/s)로 측정됩니다. 이 용량은 삽입 작업을 수행하는 데 사용됩니다. 청구 단위는 시간당 100RU/s입니다. 예를 들어 100KB 항목을 작성하는 비용은 50RU/초입니다.

쓰기 작업의 경우 초당 필요한 쓰기 수를 지원하기에 충분한 용량을 프로비저닝합니다. 쓰기 작업을 수행하기 전에 포털 또는 Azure CLI를 사용하여 프로비전된 처리량을 늘린 다음 해당 작업이 완료된 후 처리량을 줄일 수 있습니다. 쓰기 기간의 처리량은 특정 데이터에 필요한 최소 처리량과 삽입 작업에 필요한 처리량의 합계입니다. 이 계산에서는 실행 중인 다른 워크로드가 없다고 가정합니다.

비용 분석 예제

컨테이너에서 처리량 값 1,000RU/s를 구성한다고 가정합니다. 총 720시간 동안 30일 동안 24시간 동안 배포됩니다.

컨테이너는 시간당 시간당 10RU/s의 10단원으로 청구됩니다. 시간당 $0.008(시간당 100RU/초당)의 10개 단위는 시간당 $0.08로 청구됩니다.

720시간 또는 7,200개 단위(100RU)의 경우 해당 월에 대해 $57.60의 요금이 청구됩니다.

스토리지는 저장된 데이터 및 인덱스에도 사용되는 각 GB에 대해 청구됩니다. 자세한 내용은 Azure Cosmos DB 가격 책정 모델을 참조하세요.

워크로드 비용을 빠르게 예측하려면 Azure Cosmos DB 용량 계산기 사용합니다.

운영 우수성

운영 우수성은 애플리케이션을 배포하고 프로덕션 환경에서 계속 실행하는 운영 프로세스를 다룹니다. 자세한 내용은 운영 우수성대한 디자인 검토 검사 목록을 참조하세요.

모니터링

Azure Databricks는 Apache Spark를 기반으로 합니다. Azure Databricks와 Apache Spark는 모두 Apache Log4j 로깅을 위한 표준 라이브러리로 사용합니다. Apache Spark에서 제공하는 기본 로깅 외에도 Log Analytics에서 로깅을 구현할 수 있습니다. 자세한 내용은 Azure Databricks 모니터링을 참조하세요.

com.microsoft.pnp.TaxiCabReader 클래스가 승차 및 요금 메시지를 처리하므로 메시지 형식이 잘못되어 유효하지 않을 수 있습니다. 프로덕션 환경에서는 이러한 잘못된 형식의 메시지를 분석하여 데이터 손실을 방지하기 위해 신속하게 해결할 수 있도록 데이터 원본의 문제를 식별하는 것이 중요합니다. com.microsoft.pnp.TaxiCabReader 클래스는 잘못된 요금 레코드 및 승차 레코드 수를 추적하는 Apache Spark Accumulator를 등록합니다.

@transient val appMetrics = new AppMetrics(spark.sparkContext)
appMetrics.registerGauge("metrics.malformedrides", AppAccumulators.getRideInstance(spark.sparkContext))
appMetrics.registerGauge("metrics.malformedfares", AppAccumulators.getFareInstance(spark.sparkContext))
SparkEnv.get.metricsSystem.registerSource(appMetrics)

Apache Spark는 Dropwizard 라이브러리를 사용하여 메트릭을 보냅니다. 일부 네이티브 Dropwizard 메트릭 필드는 Log Analytics와 호환되지 않습니다. 따라서 이 참조 아키텍처에는 사용자 지정 Dropwizard 싱크 및 보고자가 포함됩니다. Log Analytics에서 예상하는 형식으로 메트릭의 형식을 지정합니다. Apache Spark가 메트릭을 보고할 때 잘못된 승객 및 요금 데이터에 대한 사용자 지정 메트릭도 전송됩니다.

Log Analytics 작업 영역에서 다음 예제 쿼리를 사용하여 스트리밍 작업의 작업을 모니터링할 수 있습니다. 각 쿼리의 인수 ago(1d) 마지막 날에 생성된 모든 레코드를 반환합니다. 이 매개 변수를 조정하여 다른 기간을 볼 수 있습니다.

스트림 쿼리 작업 중에 기록된 예외

SparkLoggingEvent_CL
| where TimeGenerated > ago(1d)
| where Level == "ERROR"

잘못된 형식의 요금 및 승객 데이터의 누적

SparkMetric_CL
| where TimeGenerated > ago(1d)
| where name_s contains "metrics.malformedrides"
| project value_d, TimeGenerated, applicationId_s
| render timechart

SparkMetric_CL
| where TimeGenerated > ago(1d)
| where name_s contains "metrics.malformedfares"
| project value_d, TimeGenerated, applicationId_s
| render timechart

시간에 따른 작업 작업

SparkMetric_CL
| where TimeGenerated > ago(1d)
| where name_s contains "driver.DAGScheduler.job.allJobs"
| project value_d, TimeGenerated, applicationId_s
| render timechart

리소스 조직 및 배포

프로덕션, 개발 및 테스트 환경에 대해 별도의 리소스 그룹을 만듭니다. 별도의 리소스 그룹을 만들면 배포 관리, 테스트 배포 삭제, 액세스 권한 할당 등이 더 간단해집니다.
Azure Resource Manager 템플릿 사용하여 코드로서의 인프라 프로세스에 따라 Azure 리소스를 배포합니다. 템플릿을 사용하면 Azure DevOps 서비스 또는 기타 CI/CD(연속 통합 및 지속적인 업데이트) 솔루션을 사용하여 배포를 쉽게 자동화할 수 있습니다.
각 워크로드를 별도의 배포 템플릿에 배치하고 리소스를 소스 제어 시스템에 저장합니다. 템플릿을 CI/CD 프로세스의 일부로 함께 또는 개별적으로 배포할 수 있습니다. 이 방법은 자동화 프로세스를 간소화합니다.

이 아키텍처에서 Event Hubs, Log Analytics 및 Azure Cosmos DB는 단일 워크로드로 식별됩니다. 이러한 리소스는 단일 Azure Resource Manager 템플릿에 포함됩니다.
워크로드를 준비하는 것이 좋습니다. 다음 단계로 이동하기 전에 다양한 단계에 배포하고 각 단계에서 유효성 검사를 실행합니다. 이렇게 하면 프로덕션 환경에 업데이트를 푸시하는 방법을 제어하고 예기치 않은 배포 문제를 최소화할 수 있습니다.

이 아키텍처에는 여러 배포 단계가 있습니다. Azure DevOps 파이프라인을 만들고 해당 단계를 추가하는 것이 좋습니다. 다음 단계를 자동화할 수 있습니다.
- Databricks 클러스터를 시작합니다.
- Databricks CLI를 구성합니다.
- Scala 도구를 설치합니다.
- Databricks 비밀을 추가합니다.
Databricks 코드 및 수명 주기의 품질과 안정성을 향상시키기 위해 자동화된 통합 테스트를 작성하는 것이 좋습니다.

시나리오 배포

참조 구현을 배포하고 실행하려면 GitHub 추가 정보단계를 수행합니다.

다음 단계

Azure Stream Analytics를 사용하는 스트림 처리

다음을 통해 공유