有些既存的專案把一部分日誌資訊寫入到資料庫中了，或者是由於其它的原因我們希望把關係型資料庫中的資訊讀取到 elasticsearch 中。這種情況可以使用 logstash 的 jdbc input 外掛從關係型資料庫中讀取日誌資料，然後輸出到 elasticsearch 中。本文介紹如何在 windows 系統中配置 logstash 從 SQL Server 資料庫中讀取資料。
說明：演示的環境為 windows server 2016，logstash 的版本為 6.2.4。

關鍵步驟

本文將按照下面的順序介紹使用 logstash 從 SQL Server 資料庫匯出資料的關鍵步驟：

• 安裝 Java Development Kit(JDK)
• 安裝 Logstash
• 安裝 SQL Server 的 JDBC 驅動
• 配置 Logstash
• 整合域認證
• 持續讀取資料
• 把時間戳設定為記錄產生的時間

安裝 Java Development Kit(JDK)

執行 logstash 6.2.4 需要先在環境中安裝 JDK，請不要安裝最新版本的 JDK，最好是安裝 JDK8，演示中筆者安裝的版本為 jdk-8u111-windows-x64，直接安裝到預設的目錄中。
在 logstash 的執行指令碼中用到了 JAVA_HOME 環境變數，因此我們需要先新增這個環境變數(注意，環境變數的值為 JDK 的實際安裝目錄)：

環境變數新增完成後，新啟動一個 PowerShell 視窗，執行下面的命令：

> echo $env:JAVA_HOME

通過輸出的結果驗證環境變數是否被正確新增。

安裝 Logstash

請從官方下載 logstash 的 windows 安裝包，其實就是一個 zip 檔案，比如：logstash-6.2.4.zip。Logstash 的安裝非常簡單，直接解壓縮就可以了。示例中，我把它解壓到了 C 盤的根目錄下，並重命名為 logstash，因此 logstash 的安裝目錄為：C:\logstash。
如果你想把 logstash 配置為 windows service 執行在後臺，請參考《Windows 下配置 Logstash 為後臺服務》一文。

安裝 SQL Server 的 JDBC 驅動

Logstash 需要使用 JDBC 驅動從 SQL Server 資料庫中讀取資料，因此我們還需要安裝 JDBC 驅動。同樣不要去獲取最新版本的驅動程式，請選擇 Microsoft JDBC Driver 4.2 for SQL Server。下載安裝包 sqljdbc_4.2.8112.200_enu.exe，然後執行它。其實它只是個自解壓的壓縮包，選個目錄並解壓縮。筆者選擇的 C 盤的根目錄，所以驅動檔案的絕對路徑為：
C:\sqljdbc_4.2\enu\jre8\sqljdbc42.jar

在 logstash jdbc 外掛中，我們可以直接指定這個檔案的絕對路徑，比如：

input {
  jdbc {
    jdbc_driver_library => "C:\sqljdbc_4.2\enu\jre8\sqljdbc42.jar"
    ...
  }
}

除了這種方式，我們還可以通過新增環境變數的方法來指定 JDBC 驅動。在 windows 系統中再新增一個名為 CLASSPATH 環境變數，變數的值為：

.;C:\sqljdbc_4.2\enu\jre8\sqljdbc42.jar
. 表示在當前目錄下查詢，接著是一個 ;(分號)和後面的絕對路徑。如果這個路徑中包含空格，需要使用雙引號包裹這個路徑。
在添加了環境變數 CLASSPATH 後，就可以刪除上面配置檔案中的 jdbc_driver_library 資訊了！本文的示例中將會使用 CLASSPATH 環境變數。

配置 Logstash

從 SQL Server 資料庫中讀取資料是由 logstash 的 JDBC 外掛實現的，該外掛作為 input 外掛預設已隨 logstash 安裝，可以直接使用：

input {
  jdbc {
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://DBSVR_NAME;databaseName=DB_NAME;user=****;password=****;"
    jdbc_user => "****"
    jdbc_password => "****"
    statement => "SELECT * FROM [DB].[SCHEMA].[TABLE]"
  }
}
output {
    file {
        path => "c:\output.txt"
    }
}

jdbc_connection_string 描述了到 SQL Server 的連線字串，你需要指定 SQL Server 伺服器的地址、目標資料庫的名稱以及使用者名稱稱及其密碼。jdbc_user 和 jdbc_password 是對連線字串中使用者名稱、密碼的重複。statement 則用來指定查詢語句，它返回的結果會被 logstash 獲取到。簡單起見，我把結果輸出到了本機的 c:\output.txt 檔案中，這樣比較容易除錯。
請根據你的實際情況更新上面的配置檔案，並儲存到 C:\logstash\sql.conf 檔案中，然後以管理員許可權啟動 PowerShell 並進入到 C:\logstash 目錄，執行下面的命令：

> .\logstash.bat -f .\sql.conf

如果配置資訊正確， statement 指令指定的 SQL 語句的執行結果就會被儲存到 C:\output.txt 檔案中。

整合域認證

Windows 平臺下很多場景中都會使用整合域認證的方式進行身份認證，比如在上例中採用整合域認證的方式代替連線字串中的使用者名稱和密碼：

input {
  jdbc {
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://DBSVR_NAME;databaseName=DB_NAME;integratedSecurity=true;"
    jdbc_user => ""
    statement => "SELECT * FROM [DB].[SCHEMA].[TABLE]"
  }
}

在 jdbc_connection_string 字串中我們用 integratedSecurity=true 替換了使用者名稱和密碼，並且刪除了配置項 jdbc_password。 配置項 jdbc_user 也被設定成了空字串，因為此時 jdbc_user 的值可以隨便設定，但不能不設定。

配置 sqljdbc_auth.dll
如果此時啟動 logstash 會收到 "無法載入 sqljdbc_auth.dll" 的錯誤。原因是使用域整合認證時，需要載入 sqljdbc_auth.dll，預設的設定無法找到這個 dll。這個 dll 就在我們安裝的 JDBC 驅動目錄下，我們需要在 C:\logstash\config\jvm.options  檔案中顯示指定它的路徑。比如新增下面的行：

-Djava.library.path=C:\sqljdbc_4.2\enu\auth\x64

因為我們的演示環境是 x64 架構的，所以這裡指定 x64 目錄，對於 x86 架構的系統，請指定 x86 目錄。
現在就可以通過域認證的方式訪問 SQL Server 了，重新執行一遍前面的命令試試！

持續讀取資料

使用現在的配置，每執行一遍 .\logstash.bat -f .\sql.conf 命令就會把資料重複追加到 output.txt 檔案中一遍。也就是說，每次執行 statement 語句返回的結果都基本一樣，不僅無法持續地從資料庫讀取資料，還會重複輸出已經獲取過的內容。我們可以使用 jdbc 外掛的內建變數 sql_last_value 和配置項 schedule、use_column_value、tracking_column 解決這個問題：

schedule => "* * * * *"
statement => "SELECT * FROM [DB].[SCHEMA].[TABLE] WHERE id > :sql_last_value"
use_column_value => true
tracking_column => "id"

schedule => "* * * * *" 表示每隔一分鐘重複執行一次資料讀取的操作，它支援 crontab 的語法，所以我們可以根據需要靈活設定讀取資料的間隔。內建變數 sql_last_value 會在本地儲存一個值，它記錄了上次讀取的最後一條記錄中的一個值，如果 use_column_value 被設定為 true 且 tracking_column 被設定為 "id"，則 sql_last_value 儲存的就是 id 列的最後一個值(在關係型資料庫中，id 列是比較常見的設計)。
上面配置的含義為：

• sql_last_value 變數總是記錄上次讀取的最後一條記錄中的 id。
• 每隔一分鐘執行一次資料讀取操作。
• 每次只讀取上次讀取後新增的資料。

把時間戳設定為記錄產生的時間

在日誌的查詢操作中，很多行為是基於預設的 @timestamp 欄位的。@timestamp 欄位可以簡單的理解為日誌記錄產生的時刻。但是如果我們的日誌記錄是從資料庫或其它地方匯入過來的，@timestamp 欄位預設記錄的是匯入日誌的時刻，這是不正確的。如果原有的日誌記錄中儲存有日誌產生的時刻，我們就可以由它來獲得 @timestamp 欄位的值：

filter {
    mutate {
      add_field => { "logtime" => "%{actiondatetime}" }
    }
    date {
        match => ["logtime", "ISO8601"]
        target => "@timestamp"
        remove_field => [ "logtime" ]
    }
}

上面的配置假設資料庫中 actiondatetime 列儲存了 datetime 型別的資料，通過一個臨時欄位 logtime 把 actiondatetime 列儲存的資訊設定給 @timestamp。這樣匯入後的日誌記錄的 @timestamp 欄位與 actiondatetime 欄位保持一致。

參考：
Jdbc input plugin
How to copy SQL Server data to Elasticsearch using LogStash