使用Apache SeaTunnel高效集成和管理SftpFile數據源

本文為Apache SeaTunnel已經支持的SftpFile Source Connector使用文檔，旨在幫助讀者理解如何高效地使用SFTP文件源連接器，以便輕鬆地使用Apache SeaTunnel集成和管理您的SftpFil數據源。

SftpFile 是指通過 SFTP（Secure File Transfer Protocol）協議進行文件操作的對象或組件。在網路編程和數據集成中，SFTPFile 通常用來表示和操作存儲在遠程 SFTP 伺服器上的文件。SFTP 是一種安全的文件傳輸協議，基於 SSH（Secure Shell）協議，提供了加密的文件傳輸和遠程文件操作功能。

支持的引擎

Spark

Flink

SeaTunnel Zeta

主要特性

• 批處理
• 列投影
• 並行處理
• 文件格式類型
  • 文本
  • CSV
  • JSON
  • Excel

描述

從 SFTP 文件伺服器讀取數據。

支持的數據源信息

使用 SftpFile 連接器，需要以下依賴項。可以通過 install-plugin.sh 下載，也可以從 Maven 中央倉庫獲取。

• 提示

如果你使用的是 Spark/Flink，請確保 Spark/Flink 集群已經集成了 Hadoop。Hadoop 2.x 版本已通過測試。

如果使用 SeaTunnel 引擎，安裝 SeaTunnel 引擎時會自動集成 Hadoop JAR 包。可以在 ${SEATUNNEL_HOME}/lib 目錄下檢查這個 JAR 包是否存在。

為了支持更多的文件類型，我們做了一些妥協，所以在內部訪問 Sftp 時我們使用了 HDFS 協議，這個連接器需要一些 Hadoop 依賴項，且僅支持 Hadoop 版2.9.X+ 版本。

數據類型映射

文件沒有特定的類型列表，我們可以通過在配置中指定模式來指示要將哪個 SeaTunnel 數據類型轉換為相應的數據。

Source選項

file_format_type [字元串]

支持以下文件類型：
text csv parquet orc json excel
如果將文件類型指定為 json，需要配置 Schema 模式選項，向連接器說明如何解析數據為你需要所需的 Row。
例如：上游數據如下：

{"code":  200, "data":  "get success", "success":  true}

也可以將多個數據保存在一個文件中，並通過換行符進行分隔：

{"code":  200, "data":  "get success", "success":  true}
{"code":  300, "data":  "get failed", "success":  false}

需要按照以下方式配置 Schema：

schema {
    fields {
        code = int
        data = string
        success = boolean
    }
}

連接器將生成以下數據：

例如，上游數據如下：

tyrantlucifer#26#male

如果不配置 Schema，Connector 將這樣處理上游數據：

如果分配數據模式，則除了 CSV 文件類型外，還應分配選項 delimiter。

需要配置 Schema 和分隔符如下：

delimiter = "#"
schema {
    fields {
        name = string
        age = int
        gender = string 
    }
}

連接器將生成以下數據：

Schema [配置項]

fields [配置項]
上游數據的 Schema。

如何創建 Sftp 數據同步任務

以下示例演示瞭如何創建一個數據同步任務，從 Sftp 讀取數據併在本地客戶端上列印出來：

# 設置要執行的任務的基本配置
env {
  execution.parallelism = 1
  job.mode = "BATCH"
}

# 創建連接到 Sftp 的源
source {
  SftpFile {
    host = "sftp"
    port = 22
    user = seatunnel
    password = pass
    path = "tmp/seatunnel/read/json"
    file_format_type = "json"
    result_table_name = "sftp"
    schema = {
      fields {
        c_map = "map<string, string>"
        c_array = "array<int>"
        c_string = string
        c_boolean = boolean
        c_tinyint = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_bytes = bytes
        c_date = date
        c_decimal = "decimal(38, 18)"
        c_timestamp = timestamp
        c_row = {
          C_MAP = "map<string, string>"
          C_ARRAY = "array<int>"
          C_STRING = string
          C_BOOLEAN = boolean
          C_TINYINT = tinyint
          C_SMALLINT = smallint
          C_INT = int
          C_BIGINT = bigint
          C_FLOAT = float
          C_DOUBLE = double
          C_BYTES = bytes
          C_DATE = date
          C_DECIMAL = "decimal(38, 18)"
          C_TIMESTAMP = timestamp
        }
      }
    }
  }
}

# 控制台列印讀取的 Sftp 數據
sink {
  Console {
    parallelism = 1
  }
}

本文由 白鯨開源 提供發佈支持！