Как отправить файл .pem в Dask Cluster? - PullRequest
Новая
       99

Как отправить файл .pem в Dask Cluster?

0 голосов
/ 27 апреля 2020

У меня есть следующее выражение dask, где я пытаюсь выполнить запрос sqlalchemy распределенным способом. Однако он ссылается на файл ключа .pem, который вводится в параметре connect_args. Как загрузить этот ключевой файл в кластер / рабочих dask, чтобы он позволил мне выполнить этот запрос sqlalchemy? 

def execute_query(q):
    conn = create_engine(f'presto://{user}:{password}@{host}:{port}/{catalog}/{schema}',
                               connect_args={'protocol': 'https',
                                             'requests_kwargs': {'verify': key}})
    return pd.read_sql(q, conn)

df = dd.from_delayed([
    delayed(execute_query)(q) for q in queries])

Я пытался использовать client.upload_file для отправки локального файла в кластер, но жалуется, что не может найти путь к ключу .pem 

OSError: Could not find a suitable TLS CA certificate bundle, invalid path: hdsj1ptc001.pem

1 Ответ

1 голос
/ 27 апреля 2020

Хотя Dask может обрабатывать некоторые файлы для вас (см. client.upload_file), вы должны использовать свои собственные методы для распределения конфиденциальных файлов, таких как учетные данные, в определенные c местоположения рабочего файла. система. Опции включают scp, секреты kubernetes и многие другие методы.

Если вы уверены в безопасности своего кластера, вы можете включить файл ключа в аргументы своей функции и записать его в файл в функции (см. ниже) или, если вызов позволяет, передать байты напрямую. 

def execute_query(q, key):
    if not os.path.exists(keyfile):   # if the data needs to be in a file
        open(keyfile, 'wb').write(key)
    conn = create_engine(f'presto://{user}:{password}@{host}:{port}/{catalog}/{schema}',
                               connect_args={'protocol': 'https',
                                             'requests_kwargs': {'verify': keyfile}})
    return pd.read_sql(q, conn)

key = dask.delayed(open('keyfile.pem', 'rb').read())
df = dd.from_delayed([
    delayed(execute_query)(q, key) for q in queries])
...