2. 程式人生 > 其它 >mysql與es同步的其他方案logstash

mysql與es同步的其他方案logstash

阿新 • • 發佈：2021-07-21

商品搜尋與商品詳情

課程內容

  1. mysql與es同步的其他方案
  2. laravel封裝elasticsearchService

1. mysql與es同步的其他方案

1.1 mysql與es同步的其他方案

logstash介紹

Logstash 是免費且開放的伺服器端資料處理管道，能夠從多個來源採集資料，轉換資料，然後將資料傳送到您最喜歡的“儲存庫”中。

Logstash是一個功能強大的工具，可與各種部署整合。 它提供了大量外掛，可幫助你解析，豐富，轉換和緩衝來自各種來源的資料。 如果你的資料需要 Beats 中沒有的其他處理，則需要將 Logstash 新增到部署中。

應用場景

  1. 日誌搜尋器： logstash採集、處理、轉發到elasticsearch儲存，在kibana進行展示
  2. Elk日誌分析（elasticsearch+logstash+kibana）
  3. logstash同步mysql資料庫資料到es

logstash安裝

  1. 拉取logstash映象
docker pull logstash:7.12.1(需要與es版本對應)
  1. 構建logstash容器
mkdir /docker/logstash        --建立一個用於儲存logstash配置以及外掛的目錄
docker run -p 9900:9900 -d --name logstash -v /docker/logstash:/etc/logstash/pipeline --privileged=true logstash:7.12.1
  1. 進入容器內部安裝 jdbc 和 elasticsearch 外掛
#進入logstash容器內部
docker exec -it logstash bash

#使用logstash-plugin安裝器安裝logstash-input-jdbc外掛，改安裝器在bin目錄下 (此外掛映象新版本自帶)
logstash-plugin install logstash-input-jdbc

#安裝資料輸出到es的外掛
logstash-plugin install logstash-output-elasticsearch
  1. 下載jdbc的mysql-connection.jar包

https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.24/mysql-connector-java-8.0.24.jar
  1. 修改容器內部配置
vi config/logstash.yml       更改logstash.yml檔案

內容如下:
http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://172.17.0.7:9200" ]

vi config/pipelines.yml  更改pipelines.yml檔案

內容如下：
- pipeline.id: table1
  path.config: "/etc/logstash/pipeline/logstash.conf"

需要注意的是自己目錄是不是存在這些檔案，不要找錯地方了，一般進入容器就會是logstash的安裝目錄，ls檢視就能夠看到config目錄的

  1. 退出容器，配置檔案建立與編輯 (此處屬於全量的配置檔案)
touch /docker/logstash/logstash.conf

input {
 stdin { }
    jdbc {
        #注意mysql連線地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://172.17.0.4:3306/lmrs_2008_shops"
        jdbc_user => "root"
        jdbc_password => "root"
        #這個jar包的地址是容器內的地址
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        statement => "select a.`name`,a.long_name,a.brand_id,a.three_category_id as category_id,a.shop_id,a.price,a.status,a.sold_count,a.review_count,a.create_time,a.last_time,b.`name` as category,b.path from lmrs_products as a LEFT JOIN lmrs_product_categorys as b on a.three_category_id = b.id"
        schedule => "* * * * *"
    }
 }
 output {
     elasticsearch {
        #注意mysql連線地址一定要用ip，不能使用localhost等
        hosts => "172.17.0.7:9200"
        index => "products"
        document_type => "_doc"
        document_id => "_id"
     }
     stdout {
        codec => json_lines
    }
}

增量配置檔案如下

input {
 stdin { }
    jdbc {
        #注意mysql連線地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://192.168.63.1:3306/starsky"
        jdbc_user => "starsky"
        jdbc_password => "root"
        #資料庫重連嘗試
        connection_retry_attempts => "3"
        #資料庫連線可用校驗超時時間，預設為3600s
        jdbc_validation_timeout => "3600"
        #這個jar包的地址是容器內的地址
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        #開啟分頁查詢（預設是false）
        jdbc_paging_enabled => "true"
        #單次分頁查詢條數（預設100000，欄位較多的話，可以適當調整這個數值）
        jdbc_page_size => "50000"
        #執行的sql語句
        statement => "SELECT a.id,a.`name`,a.long_name,a.brand_id,a.three_category_id AS category_id,a.shop_id,a.price,a.`STATUS`,a.sold_count,a.review_count,a.create_time,a.last_time FROM lmrs_products AS a where a.id > :sql_last_value"
        #需要記錄查詢結果某欄位的值時，此欄位為true，否則預設tracking_colum為timestamp的值
        use_column_value => true
        #是否將欄位名轉為小寫，預設為true（如果具備序列化或者反序列化，建議設定為false）
        lowercase_column_names => false
        #需要記錄的欄位，同於增量同步，需要是資料庫欄位
        tracking_column => id
        #記錄欄位的資料型別
        tracking_column_type => numeric
        #上次資料存放位置
        record_last_run => true
        #上一個sql_last_value的存放路徑，必須在檔案中指定欄位的初始值
        last_run_metadata_path => "/etc/logstash/pipeline/products.txt"
        #是否清除last_run_metadata_path的記錄，需要增量同步這個欄位的值必須為false
        clean_run => false
        #同步的頻率（分 時 天 月 年）預設為每分鐘同步一次
        schedule => "* * * * *"
    }
 }
 output {
     elasticsearch {
        #注意mysql連線地址一定要用ip，不能使用localhost等
        hosts => "172.17.0.7:9200"
        index => "products"
        document_type => "_doc"
        document_id => "%{id}"
     }
     stdout {
        codec => json_lines
    }
}

last_run_metadata_path => "/etc/logstash/pipeline/products.txt":因為需要記錄下上次同步的資料id，所以這裡會有一個檔案進行儲存這個id，需要在logstash目錄下去建立一個txt檔案，用於儲存這個id，同時需要給予許可權。不給會出現許可權異常問題

2. laravel封裝elasticsearchService

es中的商品索引資訊如下：

PUT /products/
{
  "mappings": {
    "properties": {
      "name":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "long_name":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "brand_id":{
        "type": "integer"
      },
      "category_id":{
        "type":"integer"
      },
      "category":{
        "type": "keyword"
      },
      "category_path":{
        "type": "keyword"
      },
      "shop_id":{
        "type":"integer"
      },
      "price":{
        "type":"scaled_float",
        "scaling_factor":100
      },
      "sold_count":{
        "type":"integer"
      },
      "review_count":{
        "type":"integer"
      },
      "status":{
        "type":"integer"
      },
      "create_time" : {
          "type" : "date"
      },
      "last_time" : {
          "type" : "date"
      }
    }
  }
}
  1. 建立一個Service，用於提供控制器使用es服務
php artisan make:service ElasticsearchService.php

App/Service/ElasticsearchService.php

<?php
namespace App\Services;

use App\Models\ProductCategory;

class ElasticsearchService
{
    protected $params = [
        'index' => 'products',//索引
        'type'  => '_doc',//型別
        'body' => [
            "query" => [
                "bool" => [
                    'filter' => [],
                    'must'  => []
                ]
            ]
        ]
    ];

    /**
     * @param $size  資料量
     * @param $page   索引起始
     * @return $this
     * 搜尋分頁構建
     */
    public function paginate($size,$page)
    {
        $this->params['body']['from'] = ($page - 1) * $size;
        $this->params['body']['size'] = $size;
        return $this;
    }

    /**
     * @return $this
     * 判斷商品是否已上架並經過稽核
     */
    public function IsStatus()
    {
        $this->params['body']['query']['bool']['filter'][] = ['term' => ['status' => 1]];
        return $this;
    }

    /**
     * @param ProductCategory $category 使用者傳遞過來的分類物件或者id
     * @return $this
     * 分類篩選
     */
    public function category(ProductCategory $category)
    {
        if ($category->is_directory){
            $this->params['body']['query']['bool']['filter'] = [
                'prefix' => ['category_path' => $category->path.$category->id.'-']
            ];
        }else{
            $this->params['body']['query']['bool']['filter'][] = ['term' => ['category_id' => $category->id]];
        }

        return $this;
    }

    /**
     * @param $keywords 關鍵詞陣列
     * @return $this
     * 關鍵詞按照權重進行搜尋
     */
    public function keywords($keywords)
    {
        //如果不是陣列需要轉為陣列
        $keywords = is_array($keywords) ? $keywords : [$keywords];

        foreach ($keywords as $keyword){
            $this->params['body']['query']['bool']['must'][] = [
                'multi_match' => [
                    'query' => $keyword,
                    'fields' => [
                        'long_name^3',
                        'category^2'
                    ]
                ]
            ];
        }

        return $this;
    }

    /*
     * 排序
     */
    public function orderBy($filed,$direction)
    {
        if (!isset($this->params['body']['sort'])){
            $this->params['body']['sort'] = [];
        }
        $this->params['body']['sort'][] = [$filed => $direction];
        return $this;
    }

    /*
     * 返回結構體
     */
    public function getParams()
    {
        return $this->params;
    }
}
?>

app/Http/Controllers/Api/V1/ProductController.php

<?php

namespace App\Http\Controllers\Api\V1;

use App\Http\Controllers\Controller;
use Illuminate\Http\Request;
use App\Models\ProductCategory;
use App\Models\Product;
use App\Services\ElasticsearchService;

class ProductController extends Controller
{
    /**
     * [index description]
     * @method index
     * @param  {[type]} Request [description]
     * @return {[type]}         [description]
     * 商品列表按照輸入條件搜尋
     */
    public function index(Request $request)
    {
        //分頁的起始
        $page = $request->input('page',1);
        //分頁的資料數量
        $perPage = 20;

        //呼叫es封裝類，增加商品狀態為上架條件與分頁查詢結構
        $builder = (new ElasticsearchService())->IsStatus()->paginate($perPage,$page);

        //分類搜尋
        if ($request->input('category_id') && $category = $this->category($request->input('category_id')) ){
            $builder->category($category);
        }

        //具備關鍵詞，按照關鍵詞進行搜尋（可以是多個）
        if ($search = $request->input('search','')){
            $keywords = array_filter(explode(' ',$search));
            $builder->keywords($keywords);
        }

        //根據銷量，價格，評論數量進行排序
        if ($order = $request->input('order','')){
            if (preg_match('/^(.+)_(asc|desc)$/',$order,$m)){
                if (in_array($m[1],['price','sold_count','review_count'])){
                    $builder->orderBy($m[1],$m[2]);
                }
            }
        }

        //通過容器註冊的es單例呼叫search方法到es搜尋資料，條件為上面構建的結構體
        $restful = app('es')->search($builder->getParams());

        return response()->json([
            "data" => $restful
        ]);
    }

    /**
     * [category description]
     * @method category
     * @param  {[type]} $category [description]
     * @return {[type]}           [description]
     * 查詢分類資料
     */
    public function category($category)
    {
        $category_array = explode(',',$category);
        $category_id = array_pop($category_array);
        return ProductCategory::query()->where('id',$category_id)->first();
    }
}

?>

相關推薦

mysql與es同步的其他方案logstash

商品搜尋與商品詳情 課程內容 mysql與es同步的其他方案 laravel封裝elasticsearchService

binlog redis 驗證訂單_Canal 詳解 Mysql與Redis資料同步 解決方案

技術標籤：binlog redis 驗證訂單 要保證資料庫和redis強一致性是不可能的，肯定有少許時間的不一致。canal是阿里的一套元件，用來監聽mysql master傳送的類似binary log的資料，然後讓訊息費去消費。

Centos7系統下Mysql主從同步配置方案

前言 最近在做專案高可用時，需要使用資料同步。由於只有雙節點，且採用主主同步可能存在迴圈同步的風險，故綜合考慮採用Mysql主從同步（Master-Slave同步）。

詳解MySQL雙活同步複製四種解決方案

對於資料實時同步，其核心是需要基於日誌來實現，是可以實現準實時的資料同步，基於日誌實現不會要求資料庫本身在設計和實現中帶來任何額外的約束。

阿里雲RDS與ECS自建mysql資料庫主從同步(GTID方式)

1、GTID的概念全域性事務標識：global transaction identifiers；GTID是一個事務一一對應，並且全域性唯一ID；一個GTID在一個伺服器上只執行一次，避免重複執行導致資料混亂或者主從不一致；GTID用來代替傳統複製方法

記linux與Windows之間實現檔案實時同步的方案

近日，資料分析部提了個需求說要讓一臺linux的機器上的檔案同步到一臺Windows機器的某個目錄下，由於本姑娘從來沒有碰到過這種問題，於是上網找了下資料發現了一個很好用的工具可以幫忙，那就是cwRsyncSer

Canal+Kafka實現MySQL與Redis資料同步

思維導圖 前言 在很多業務情況下，我們都會在系統中加入redis快取做查詢優化。

全文搜尋引擎 Elasticsearch （四）MySQL如何實時同步資料到ES

canal簡介 canal主要用途是對MySQL資料庫增量日誌進行解析，提供增量資料的訂閱和消費，簡單說就是可以對MySQL的增量資料進行實時同步，支援同步到MySQL、Elasticsearch、HBase等資料儲存中去。

LogStash實現MySQL資料增量同步到ElasticSearch

需求的由來 在做一個專案的時候，需要一個搜尋的功能，剛開始想到的是用資料庫的模糊查詢，但是考慮到效率的問題，於是就用了ElasticSearch（ES），但是MySQL中的資料怎麼樣到ES中呢，我們可能會想到，MySQL

MYSQL主從複製（非同步與半同步複製）

技術標籤：運維 主從複製 原理：將主伺服器的binlog日誌複製到從伺服器上執行一遍，達到主從資料的一致狀態。**

MySQL的3節點主從同步複製方案

上篇文章《為什麼要對MySQL做主從同步複製》我們說明了MySQL主從同步的作用，主從同步的原理和主從同步的缺點。下面我們介紹下3節點中：2個節點互為主從，1個節點作為前2個節點的從，用於實現MySQL5.6的3節點主從同步

Logstash之MySQL做資料同步 報錯記錄

技術標籤：外掛元件java 1.首先說明的是 ES採用UTC 時區，比北京時間早8小時，所以ES讀取資料時讓最後更新時間加8小時（後半句錯誤說法）

6、Canal實現MySQL到ES實時同步-1

1、準備一臺虛擬機器 1.1 配置靜態ip 可以參考 https://www.cnblogs.com/braveym/p/9096402.html 

mysql資料庫主從同步的3種一致性方案實現，以及優劣比較

方法1：半同步複製 介於非同步複製和同步複製之間，主庫在執行完客戶端提交的事務後不會立即返回給客戶端，

MySQL與TiDB基礎知識類比

溫故而知新 可以為師矣 一、儲存模型 MySQL Each tablespace consists of database pages with a default size of 16KB. The pages are grouped into extents of size 1MB (64 consecutive pages). The “files”

增量同步-spring batch(6)動態引數繫結與增量同步

tags：springbatch 1.引言 上一篇《便捷的資料讀寫-spring batch(5)結合beetlSql進行資料讀寫》中使用Spring Batch及BeetlSql，對資料庫讀寫元件進行資料庫同步，實際上是全量同步。全量同步的問題在於每次需要讀取

Spring Boot Mysql 版本驅動連線池方案選擇

國內環境下，用Mysql還是比較多的。這裡簡單的總結了一下，如有繆誤，還請指正。

解決mysql與navicat建立連接出現1251錯誤

重灌了電腦，安裝了最新版的MySQL資料庫，結果Navicat連線Mysql報1251錯誤，sqlyog報2058錯誤，但是window命令進入mysql，賬號密碼都是正確的。

MySQL 整表加密解決方案 keyring_file詳解

說明 MySql社群版從5.7.11開始支援基於表的資料加密方案，模組名為keyring_file，支援加密整張表。這種是加密方式其實是基於檔案加密的，一旦mysqld讀取key啟動後，將會解密整張表的資料，在mysql服務內，讀取的資料

MySQL字符集亂碼及解決方案分享

前言 字符集是一套符號和編碼的規則，不論是在oracle資料庫還是在mysql資料庫，都存在字符集的選擇問題，而且如果在資料庫建立階段沒有正確選擇字符集，那麼可能在後期需要更換字符集，而字符集的更換是代價比較高的