Задание потока данных не использует PubSub при запуске из шаблона - PullRequest
Новая
       40

Задание потока данных не использует PubSub при запуске из шаблона

0 голосов
/ 19 октября 2018

В настоящее время у меня есть задание, которое выводит содержимое темы pubsub в папку облачного хранилища, которая отлично работает, если я запускаю jar напрямую.

Однако всякий раз, когда я пытаюсь запустить задание, используя шаблон, который язагружено, сообщения не передаются по конвейеру.

Это очень похоже на предоставленный Google шаблон , за исключением того, что он принимает подписку вместо темы.

Вот моя конфигурация: 

trait Options extends PipelineOptions with StreamingOptions {
  @Description("The Cloud Pub/Sub subscription to read from")
  @Default.String("projects/project/subscriptions/subscription")
  def getInputSubscription: String
  def setInputSubscription(value: String): Unit

  @Description("The Cloud Storage directory to output files to, ends with /")
  @Default.String("gs://tmp/")
  def getOutputDirectory: String
  def setOutputDirectory(value: String): Unit

  @Description("The Cloud Storage prefix to output files to")
  @Default.String("subscription-")
  def getOutputFilenamePrefix: String
  def setOutputFilenamePrefix(value: String): Unit

  @Description("The shard template which will be part of the filenames")
  @Default.String("-W-P-SSSSS-of-NNNNN")
  def getShardTemplate: String
  def setShardTemplate(value: String): Unit

  @Description("The suffix of the filenames written out")
  @Default.String(".txt")
  def getOutputFilenameSuffix: String
  def setOutputFilenameSuffix(value: String): Unit

  @Description("The window duration in minutes, defaults to 5")
  @Default.Integer(5)
  def getWindowDuration: Int
  def setWindowDuration(value: Int): Unit

  @Description("The compression used (gzip, bz2 or none), bz2 can't be loaded into BigQuery")
  @Default.String("none")
  def getCompression: String
  def setCompression(value: String): Unit

  @Description("The maximum number of output shards produced when writing")
  @Default.Integer(1)
  def getNumShards: Int
  def setNumShards(value: Int): Unit
}

и вот как я запускаю шаблон: 

   gcloud dataflow jobs run storage \
     --gcs-location gs://bucket/templates/Storage \
     --parameters runner=DataflowRunner,project=project,streaming=true,inputSubscription=projects/project/subscriptions/sub,outputDirectory=gs://bucket/

Вот как я запускаю работу без шаблона: 

./storage \
  --runner=DataFlowRunner \
  --project=project \
  --streaming=true \
  --gcpTempLocation=gs://tmp-bucket/ \
  --inputSubscription=projects/project/subscriptions/sub  \
  --outputDirectory=gs://bucket/

1 Ответ

0 голосов
/ 06 ноября 2018

Как сказано в комментарии @GuillemXercavins, параметры должны использовать интерфейс ValueProvider в качестве типа .Это позволит устанавливать или использовать параметры конвейера во время выполнения, что и вызвало проблему.

Стоит отметить, как вы уже сделали в комментарии, что ValueProvider кажется не поддерживается в Scio .

Редактировать:

Пример Scio предоставлено @ BenFradet в комментарииниже.

...