Проблема решен ! Не мога да повярвам, че прекарах цели два дни в това... Гледах напълно в грешната посока.
Проблемът не беше с някаква мрежова конфигурация на Dataflow или GCP и доколкото мога да преценя...
е вярно.
Проблемът, разбира се, беше в моя код:само проблемът беше разкрит само в разпределена среда. Направих грешката да отворя тунела от главния тръбопроводен процесор, вместо от работниците. Така че SSH тунелът беше изграден, но не между работниците и целевия сървър, а само между главния конвейер и целта!
За да поправя това, трябваше да променя моя искащ DoFn, за да обвия изпълнението на заявката с тунела:
class TunnelledSQLSourceDoFn(sql.SQLSourceDoFn):
"""Wraps SQLSourceDoFn in a ssh tunnel"""
def __init__(self, *args, **kwargs):
self.dbport = kwargs["port"]
self.dbhost = kwargs["host"]
self.args = args
self.kwargs = kwargs
super().__init__(*args, **kwargs)
def process(self, query, *args, **kwargs):
# Remote side of the SSH Tunnel
remote_address = (self.dbhost, self.dbport)
ssh_tunnel = (self.kwargs['ssh_host'], self.kwargs['ssh_port'])
with open_tunnel(
ssh_tunnel,
ssh_username=self.kwargs["ssh_user"],
ssh_password=self.kwargs["ssh_password"],
remote_bind_address=remote_address,
set_keepalive=10.0
) as tunnel:
forwarded_port = tunnel.local_bind_port
self.kwargs["port"] = forwarded_port
source = sql.SQLSource(*self.args, **self.kwargs)
sql.SQLSouceInput._build_value(source, source.runtime_params)
logging.info("Processing - {}".format(query))
for records, schema in source.client.read(query):
for row in records:
yield source.client.row_as_dict(row, schema)
както виждате, трябваше да отменя някои части от библиотеката pysql_beam.
И накрая, всеки работник отваря свой собствен тунел за всяка заявка. Вероятно е възможно да се оптимизира това поведение, но е достатъчно за моите нужди.
