Прием файлов с серверов SFTP

Узнайте, как принять файлы с серверов SFTP с помощью Lakeflow Connect. Соединитель SFTP расширяет функциональные возможности автозагрузчика, обеспечивая безопасный и инкрементный прием данных с серверов SFTP с управлением единым каталогом Unity.

Ключевые особенности

Соединитель SFTP предлагает следующее:

  • Закрытый ключ и проверка подлинности на основе пароля.
  • Инкрементное поглощение и обработка файлов с гарантиями точного выполнения.
  • Автоматическое вывод схемы, эволюция и спасение данных.
  • Управление каталогом Unity для безопасной загрузки и аутентификационных данных.
  • Поддержка расширенных форматов файлов: JSON, CSV, XMLPARQUET, , AVRO, TEXTBINARYFILEи ORC.
  • Встроенная поддержка сопоставления шаблонов и подстановочных знаков для простого определения подмножеств данных.
  • Доступно во всех типах вычислительных ресурсов, включая конвейеры Lakeflow, Databricks SQL, бессерверные и классические вычислительные ресурсы, начиная с Databricks Runtime 17.3.

Перед тем как начать

Чтобы создать подключение и конвейер приема, необходимо следующее:

  • Рабочая область с включённым каталогом Unity.
  • CREATE CONNECTION Привилегии для создания подключения SFTP или соответствующие привилегии для использования существующего подключения в зависимости от режима доступа к кластеру:
    • Выделенный режим доступа: MANAGE CONNECTION.
    • Стандартный режим доступа: USE CONNECTION.
  • Вычисление, использующее Databricks Runtime версии 17.3 или более поздней. Для рабочих пространств, соответствующих требованиям FedRAMP, Canada Protected B и IRAP, требуется Databricks Runtime 18 LTS или выше.

Настройка SFTP

Сначала убедитесь, что исходный сервер SFTP доступен для среды кластера Databricks:

  • Убедитесь, что удаленный сервер доступен в VPC, настроенном для рабочей области.
  • Убедитесь, что правила SSH разрешают диапазон IP-адресов VPC Databricks (если вы используете классические вычисления) или исходящие IP-адреса (если вы используете бессерверные вычисления).
  • На классической вычислительной плоскости настройте стабильный IP-адрес с подсистемой балансировки нагрузки, шлюзом NAT, интернет-шлюзом или эквивалентом и подключите его к подсети, где развертываются Azure Databricks вычислений. Это позволяет вычислительному ресурсу совместно использовать стабильный общедоступный IP-адрес, который можно разрешить на стороне сервера SFTP. Инструкции по настройке параметров сети см. в разделе "Одноранговые виртуальные сети".
  • Для плоскости бессерверных вычислений см. раздел Настройка частного подключения к ресурсам Azure, чтобы получить исходящие IP-адреса.

Создание подключения

Создайте подключение каталога Unity для хранения учетных данных SFTP. У вас должны быть CREATE CONNECTION привилегии.

Соединитель поддерживает следующие методы проверки подлинности:

  • Закрытый ключ PEM
  • Аутентификация на основе пароля

Databricks рекомендует использовать проверку подлинности закрытого ключа PEM. Databricks также рекомендует использовать учетные данные с наименьшими привилегиями на исходном сервере SFTP (например, не корневой пользователь, ограниченный доступом только для чтения).

При создании потока данных соединитель пытается автоматически найти подключение, которое можно использовать и которое соответствует хосту. При наличии нескольких сопоставленных подключений соединитель выбирает первое подключение, которое успешно подключается к узлу. Начиная с DBR 18.2, можно явно указать подключение с помощью databricks.connection параметра. Это рекомендуемый вариант и избегает неоднозначности при наличии нескольких подключений для одного узла.

Обозреватель каталогов

  1. В рабочей области Azure Databricks щелкните значок данных. Каталог.

  2. Щелкните значок Подключитесь, а затем щелкните "Подключения".

  3. Нажмите кнопку "Создать подключение ".

  4. На странице "Основы подключения " мастера настройки подключения введите уникальное имя подключения.

  5. Для типа подключения выберите SFTP.

  6. Для типа проверки подлинности выберите закрытый ключ PEM.

  7. Нажмите кнопку Далее.

  8. На странице проверки подлинности, для Host, введите имя узла внешнего сервера.

  9. Для пользователя введите удостоверение пользователя, используемое для доступа к внешнему экземпляру.

  10. Нажмите кнопку Далее.

  11. На странице сведений о подключении введите закрытый ключ в формате PEM. Также укажите парольную фразу ключа, если применимо.

  12. Если вы хотите пропустить проверку отпечатка ключа узла, отключите настройку Обязательный отпечаток ключа узла.

    Если этот параметр выбран, подключение продолжается только в том случае, если открытый ключ сервера соответствует ожидаемому отпечатку sha-256. При отключении подключение продолжается независимо от соответствия. Прежде чем отключить эту функцию, обратитесь к администратору сети.

  13. Если установлено Enforce host key fingerprint, введите отпечаток сервера SFTP.

    Вы можете получить отпечатки пальцев от администратора сервера или с помощью команд CLI. Вы также можете нажать клавишу Test и создать тест подключения>. Полученное сообщение об ошибке предоставляет отпечатки пальцев. Рассмотрим пример.

    ECDSA key fingerprint is SHA256:XXX/YYY

  14. Выберите "Тестировать" и создайте подключение.

  15. Если подключение выполнено успешно, нажмите кнопку "Создать".

SQL

-- Create a connection using a username and SSH private key.
CREATE CONNECTION my_sftp_connection
TYPE sftp
OPTIONS (
  host 'my.sftpserver.com',
  -- The following credentials can also be used in-line, but Databricks recommends
  -- accessing them using the secrets scope.
  user secret('my_secret_scope','my_sftp_username'),
  pem_private_key secret('my_secret_scope','my_sftp_private_key'),
  -- Port for the host
  port '22',
  -- Passphrase for the private key (optional).
  pem_key_passphrase secret('my_secret_scope','my_sftp_private_key_passphrase'),
  -- SFTP server fingerprint. You can retrieve this from your server administrator or using CLI commands.
  key_fingerprint 'SHA256:ASampleFingerprintValueZy...',
);

Аутентификация на основе пароля

Обозреватель каталогов

  1. В рабочей области Azure Databricks щелкните значок данных. Каталог.

  2. Щелкните значок Подключитесь, а затем щелкните "Подключения".

  3. Нажмите кнопку "Создать подключение ".

  4. На странице "Основы подключения " мастера настройки подключения введите уникальное имя подключения.

  5. Для типа подключения выберите SFTP.

  6. Для типа проверки подлинности выберите имя пользователя и пароль.

  7. Нажмите кнопку Далее.

  8. На странице проверки подлинности, для Host, введите имя узла внешнего сервера.

  9. Для пользователя введите удостоверение пользователя, используемое для доступа к внешнему экземпляру.

  10. В поле "Пароль" введите пароль внешнего экземпляра.

  11. Нажмите кнопку Далее.

  12. Если вы хотите пропустить проверку отпечатка ключа узла, отключите настройку Обязательный отпечаток ключа узла.

    Если этот параметр выбран, подключение продолжается только в том случае, если открытый ключ сервера соответствует ожидаемому отпечатку sha-256. При отключении подключение продолжается независимо от соответствия. Прежде чем отключить эту функцию, обратитесь к администратору сети.

  13. Если установлено Enforce host key fingerprint, введите отпечаток сервера SFTP.

    Вы можете получить отпечатки пальцев от администратора сервера или с помощью команд CLI. Вы также можете нажать клавишу Test и создать тест подключения>. Полученное сообщение об ошибке предоставляет отпечатки пальцев. Рассмотрим пример.

    ECDSA key fingerprint is SHA256:XXX/YYY

  14. Выберите "Тестировать" и создайте подключение.

  15. Если подключение выполнено успешно, нажмите кнопку "Создать".

SQL

-- Create a connection using a username and password.
CREATE CONNECTION my_sftp_connection
TYPE sftp
OPTIONS (
  host 'my.sftpserver.com',
  user secret('my_secret_scope','my_sftp_username'),
  password secret('my_secret_scope','my_sftp_password'),
  -- Port for the host.
  port '22',
  -- SFTP server fingerprint. You can retrieve this from your server administrator or using CLI commands.
  key_fingerprint 'SHA256:ASampleFingerprintValueZy...',
);

Чтение файлов с сервера SFTP

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

Вы можете указать подключение каталога Unity для проверки подлинности одним из двух способов:

  • Явный (рекомендуется, требуется DBR 18.2 и выше): Укажите подключение по имени с помощью databricks.connection параметра. При указании подключения узел и порт в URI должны соответствовать сохранённым учётным данным для подключения. Поле имени пользователя в URI является необязательным. Если не указано, соединитель использует имя пользователя из подключения.
  • Автоматически: Если не указать подключение с помощью параметра databricks.connection, соединитель определяет подключение, сопоставляя <host> и <username> в URI с доступными подключениями. При наличии нескольких сопоставленных подключений соединитель использует первый, который успешно подключается.

В следующем примере показано, как считывать файлы с помощью автоматического разрешения подключения:

# Run the Auto Loader job to ingest all existing data in the SFTP server.
# The <username> and <host> in the URI must match the connection created in the previous step.
# The connector automatically resolves the matching Unity Catalog connection for authentication.
df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.schemaLocation", "<path to store schema information>") # This is a cloud storage path
  .option("cloudFiles.format", "csv") # Or other format supported by Auto Loader
  # Specify the absolute path on the SFTP server starting from the root /.
  # Example: /home/<username>/data/files or /uploads/csv_files
  .load("sftp://<username>@<host>:<port>/<absolute_path_to_files>")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "<path to store checkpoint information>") # This is a cloud storage path.
  .trigger(availableNow = True)
  .table("<table name>"))
df.awaitTermination()

В следующем примере показано, как явно указать подключение каталога Unity с помощью databricks.connection параметра. Для этого требуется DBR 18.2 или более поздней версии.

# Requires DBR 18.2 or above. Explicitly specify the Unity Catalog connection by name (recommended).
# The username is optional in the URI when databricks.connection is provided.
df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.schemaLocation", "<path to store schema information>") # This is a cloud storage path
  .option("cloudFiles.format", "csv") # Or other format supported by Auto Loader
  .option("databricks.connection", "<connection_name>")
  # Specify the absolute path on the SFTP server starting from the root /.
  # Example: /uploads/csv_files
  .load("sftp://<host>:<port>/<absolute_path_to_files>")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "<path to store checkpoint information>") # This is a cloud storage path.
  .trigger(availableNow = True)
  .table("<table name>"))
df.awaitTermination()

В следующих примерах показано, как считывать файлы с сервера SFTP с помощью автозагрузчика в конвейерах Lakeflow:

Python

from pyspark import pipelines as dp

# The <username> and <host> in the URI must match the connection created in the previous step.
# The connector automatically resolves the matching Unity Catalog connection for authentication.
@dp.table
def sftp_bronze_table():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv") # Or other format supported by Auto Loader
    # Specify the absolute path on the SFTP server starting from the root /.
    # Example: /home/username/data/files or /uploads/csv_files
    .load("sftp://<username>@<host>:<port>/<absolute_path_to_files>")))

SQL

-- The <username> and <host> in the URI must match the connection created in the previous step.
-- The connector automatically resolves the matching Unity Catalog connection for authentication.
CREATE OR REFRESH STREAMING TABLE sftp_bronze_table
AS SELECT * FROM STREAM read_files(
  "sftp://<username>@<host>:<port>/<absolute_path_to_files>",
  format => "csv"
)

Настройка параметров автозагрузчика. Поддерживаются все параметры, кроме следующих:

  • cloudFiles.useNotifications
  • cloudFiles.useManagedFileEvents
  • cloudFiles.cleanSource
  • Параметры, относящиеся к облаку

Ограничения

  • SFTP не поддерживается в других поверхностях приема, включая COPY INTO, spark.readи dbutils.ls.
  • Обратная запись на сервер SFTP не поддерживается.
  • Автозагрузчик cleanSource (удаление или архивация файлов в источнике после приема) не поддерживается.
  • Протокол FTP не поддерживается.

Часто задаваемые вопросы

Найдите ответы на часто задаваемые вопросы о соединителе SFTP.

Как использовать подстановочные знаки или шаблоны имен файлов для выбора файлов для загрузки?

Соединитель SFTP основывается на стандартной платформе Auto Loader для работы с серверами SFTP. Это означает, что поддерживаются все параметры автозагрузчика. Для шаблонов имен файлов и подстановочных знаков используйте опции pathGlobFilter или fileNamePattern. См. функцию автозагрузчика.

Может ли соединитель SFTP получать зашифрованные файлы? (Поддерживается ли PGP?)

Соединитель не расшифровывает данные при передаче, однако зашифрованные файлы можно получать в виде двоичных и расшифровывать после приема.

Как обрабатывать несовместимые форматы закрытого ключа?

Поддерживается только формат PEM. Вы можете создать закрытый ключ в формате PEM, выполнив одно из следующих действий:

  • (Вариант 1) Создайте новый ключ RSA в стандартном формате PEM:

    % ssh-keygen -t rsa -m pem

  • (Вариант 2) Преобразуйте существующий ключ формата OpenSSH в формат PEM:

    % ssh-keygen -p -m pem -f /path/to/key  # This updates the key file.

Дополнительные ресурсы

  • Last updated on