DataFlow (PY 2.x SDk) ReadFromPubSub :: id_label & timestamp_attribute ведет себя неожиданно

Question

Мой конвейер лучей apache (использующий Python SDK + DirecrRunner для целей тестирования…) читает из темы Pubsub

Опубликованы следующие сообщения и атрибуты:

message: [{"col1": "test column 1", "col2": "test column 1"}]

attributes:{
  'event_time_v1': str(time.time()),
  'record_id': 'row-1’,
}

Я использую функцию beam.io.gcp.pubsub.ReadFromPubSub . В коде / документе упоминаются аргументы id_label и timestamp_attribute (я считаю, что это очень новые дополнения ?! Обновлено только 13 дней назад ..)

1. Когда я использую id_label для назначения каждому элементу уникального идентификатора в целях дедупликации, я получаю следующую ошибку:

NotImplementedError: DirectRunner: id_label не поддерживается для чтения PubSub````

почему так? правильно ли я понимаю, что какая-то реализация кода все еще отсутствует или я что-то здесь упускаю?

2. Когда я использую timestamp_attribute = 'event_time_v1’, чтобы назначить собственную метку времени каждому элементу (время события на стороне клиента, переданное в атрибуте сообщения event_time_v1), я замечаю, что метка времени, фактически назначенная элементу, все еще остается временем публикации сообщения

почему так? Я ожидал, что это будет время, прошедшее в event_time_v1

Я использую следующий DoFn для печати метки времени элемента

class PrintFn(beam.DoFn):

      print(element, timestamp)
      return [element]

Большое спасибо заранее за любые объяснения этому

Answer 1

У меня была та же проблема с этим сегодня, на самом деле в Jira есть открытая проблема для id_label и timestamp_attribute, недоступного в непосредственном организаторе (и я предполагаю, что при чтении все участники, не использующие поток данных).Я успешно смог использовать id_label при указании DataflowRunner в качестве бегуна (с некоторыми другими проблемами, но это кстати).

Проблема Jira ниже:

https://issues.apache.org/jira/browse/BEAM-4275?jql=text%20~%20%22python%20id_label%22

Таким образом, в настоящий момент может показаться, что это невозможно сделать с помощью прямого бегуна.