Subskrybuj usługę Google Pub/Sub
Usługa Azure Databricks udostępnia wbudowany łącznik do subskrybowania usługi Google Pub/Sub w środowisku Databricks Runtime 13.3 LTS i nowszym. Ten łącznik zapewnia semantyka przetwarzania dokładnie raz dla rekordów od subskrybenta.
Uwaga
Pub/Sub może publikować zduplikowane rekordy, a rekordy mogą zostać dostarczone do subskrybenta z zamówienia. Należy napisać kod usługi Azure Databricks, aby obsługiwać zduplikowane i out-of-order rekordy.
Przykład składni
Poniższy przykład kodu przedstawia podstawową składnię konfigurowania przesyłania strumieniowego ze strukturą odczytu z pubu/pod:
val authOptions: Map[String, String] =
Map("clientId" -> clientId,
"clientEmail" -> clientEmail,
"privateKey" -> privateKey,
"privateKeyId" -> privateKeyId)
val query = spark.readStream
.format("pubsub")
// we will create a Pubsub subscription if none exists with this id
.option("subscriptionId", "mysub") // required
.option("topicId", "mytopic") // required
.option("projectId", "myproject") // required
.options(authOptions)
.load()
Aby uzyskać więcej opcji konfiguracji, zobacz Configure options for Pub/Sub streaming read (Konfigurowanie opcji przesyłania strumieniowego pub/sub).
Konfigurowanie dostępu do usługi Pub/Sub
Usługa Databricks zaleca używanie wpisów tajnych podczas udostępniania opcji autoryzacji. Do autoryzowania połączenia wymagane są następujące opcje:
clientEmailclientIdprivateKeyprivateKeyId
W poniższej tabeli opisano role wymagane dla skonfigurowanych poświadczeń:
| Role | Wymagane lub opcjonalne | Jak jest używany |
|---|---|---|
roles/pubsub.viewer lub roles/viewer |
Wymagania | Sprawdź, czy subskrypcja istnieje i uzyskaj subskrypcję |
roles/pubsub.subscriber |
Wymagania | Pobieranie danych z subskrypcji |
roles/pubsub.editor lub roles/editor |
Opcjonalnie | Umożliwia tworzenie subskrypcji, jeśli nie istnieje, a także umożliwia korzystanie z tej deleteSubscriptionOnStreamStop opcji w celu usunięcia subskrypcji po zakończeniu przesyłania strumieniowego |
Schemat pubu/podrzędnego
Schemat strumienia jest zgodny z rekordami pobranymi z pubu/pod, zgodnie z opisem w poniższej tabeli:
| Pole | Typ |
|---|---|
messageId |
StringType |
payload |
ArrayType[ByteType] |
attributes |
StringType |
publishTimestampInMillis |
LongType |
Konfigurowanie opcji odczytu przesyłania strumieniowego pub/podrzędnego
W poniższej tabeli opisano opcje obsługiwane przez pub/sub. Wszystkie opcje są konfigurowane w ramach odczytu przesyłania strumieniowego ze strukturą przy użyciu .option("<optionName>", "<optionValue>") składni.
Uwaga
Niektóre opcje konfiguracji Pub/Sub używają koncepcji pobierania zamiast mikrosadów. Odzwierciedla to szczegóły implementacji wewnętrznej i opcje działają podobnie do rejestracji w innych łącznikach przesyłania strumieniowego ze strukturą, z tą różnicą, że rekordy są pobierane, a następnie przetwarzane.
| Opcja | Wartość domyślna | opis |
|---|---|---|
numFetchPartitions |
Ustaw na połowę liczby funkcji wykonawczych obecnych podczas inicjowania strumienia. | Liczba równoległych zadań platformy Spark, które pobierają rekordy z subskrypcji. |
deleteSubscriptionOnStreamStop |
false |
Jeśli truesubskrypcja przekazana do strumienia zostanie usunięta po zakończeniu zadania przesyłania strumieniowego. |
maxBytesPerTrigger |
Brak | Miękki limit rozmiaru partii, który ma być przetwarzany podczas każdej wyzwalanej mikrosadowej partii. |
maxRecordsPerFetch |
1000 | Liczba rekordów do pobrania na zadanie przed przetworzeniem rekordów. |
maxFetchPeriod |
10 sekund | Czas trwania każdego zadania do pobrania przed przetworzeniem rekordów. Usługa Databricks zaleca użycie wartości domyślnej. |
Semantyka przetwarzania wsadowego przyrostowego dla pubu/podrzędnego
Za pomocą Trigger.AvailableNow polecenia można korzystać z dostępnych rekordów ze źródeł Pub/Sub w partii przyrostowej.
Usługa Azure Databricks rejestruje znacznik czasu podczas rozpoczynania odczytu z ustawieniem Trigger.AvailableNow . Rekordy przetwarzane przez partię obejmują wszystkie pobrane wcześniej dane i wszystkie nowo opublikowane rekordy z sygnaturą czasową mniejszą niż znacznik czasu rozpoczęcia zarejestrowanego strumienia.
Zobacz Konfigurowanie przyrostowego przetwarzania wsadowego.
Monitorowanie metryk przesyłania strumieniowego
Metryki postępu przesyłania strumieniowego ze strukturą raportują liczbę pobranych i gotowych do przetworzenia rekordów, rozmiar pobranych i gotowych do przetworzenia rekordów oraz liczbę duplikatów widocznych od rozpoczęcia strumienia. Poniżej przedstawiono przykład tych metryk:
"metrics" : {
"numDuplicatesSinceStreamStart" : "1",
"numRecordsReadyToProcess" : "1",
"sizeOfRecordsReadyToProcess" : "8"
}
Ograniczenia
Wykonywanie spekulatywne (spark.speculation) nie jest obsługiwane w przypadku usługi Pub/Sub.