ES

Input

clean_run 是否应该保留以前的运行状态
columns_charset 特定列的字符编码。此选项将覆盖指定列的选项 columns_charset => { “column0” => “ISO-8859-1” }
connection_retry_attempts 尝试连接到数据库的最大次数(默认1)
jdbc_driver_class org.postgresql.Driver
jdbc_driver_library JDBC驱动 D:/xx.jar
jdbc_connection_string jdbc:xxx?characterEncoding=utf8&serverTimezone=UTC
jdbc_user postgres
jdbc_password
statement SELECT * from add_bridge where id = 1
parameters 查询参数的哈希 { “target_id” => “321” }
schedule 设置监听间隔 各字段含义（从左至右）分、时、天、月、年，全为*默认含义为每分钟都更新 schedule => “10 “ (10分钟执行一次)
lowercase_column_names 是否强制使用标识符字段的小写，默认 true
statement 要使用参数，请使用命名参数语法 “SELECT * FROM xx WHERE id = :target_id”
jdbc_default_timezone Asia/Shanghai 
  1. input {
  2.   jdbc {
  3.     jdbc_driver_library => "D:/code/elasticsearch/logstash-7.11.1-windows-x86_64/logstash-7.11.1/test/postgresql-42.2.19.jar"
  4.     jdbc_driver_class => "org.postgresql.Driver"
  5.     jdbc_connection_string => "jdbc:postgresql://localhost:5432/jtstest?characterEncoding=utf8&serverTimezone=GMT%2B8"
  6.     jdbc_user => "postgres"
  7.     jdbc_password => "953598751"
  8.     schedule => "* * * * *"
  9.     # statement => "SELECT * from logstash_text where times > :sql_last_value and times < now()"
  10.     statement => "SELECT * from logstash_text where times > :sql_last_value and times < now()"
  11.     tracking_column => "times"
  12.     id => "logstash_text"
  13.     use_column_value => true
  14.     tracking_column_type => "timestamp"
  15.     jdbc_default_timezone =>"Asia/Shanghai"
  16.     type => "type1"
  17.   }
  18.   jdbc {
  19.     jdbc_driver_library => "D:/code/elasticsearch/logstash-7.11.1-windows-x86_64/logstash-7.11.1/test/postgresql-42.2.19.jar"
  20.     jdbc_driver_class => "org.postgresql.Driver"
  21.     jdbc_connection_string => "jdbc:postgresql://localhost:5432/jtstest?characterEncoding=utf8&serverTimezone=GMT%2B8"
  22.     jdbc_user => "postgres"
  23.     jdbc_password => "953598751"
  24.     schedule => "* * * * *"
  25.     statement => "SELECT * from test_test where times > :sql_last_value and times < now()"
  26.     tracking_column => "times"
  27.     id => "test_test"
  28.     use_column_value => true
  29.     tracking_column_type => "timestamp"
  30.     jdbc_default_timezone =>"Asia/Shanghai"
  31.     type => "type2"
  32.   }
  33. }
  34. filter {
  35.     json {
  36.         source => "message"
  37.         remove_field => ["message"]
  38.     }
  39. }
  40. output {
  41.     if[type] == "type1" {
  42.         elasticsearch {
  43.             hosts => ["localhost:9200"]
  44.             index => "index_logstash_text"
  45.             id => "id_index_logstash_text"
  46.             document_id => "%{id}"
  47.         }
  48.     }
  49.     if[type] == "type2" {
  50.         elasticsearch {
  51.             hosts => ["localhost:9200"]
  52.             index => "index_test_test"
  53.             id => "id_index_test_test"
  54.             document_id => "%{id}"
  55.         }
  56.     }
  57.     stdout {
  58.         # codec => json_lines
  59.         codec => rubydebug 
  60.     }
  61. }

自定义分析器

  1. PUT /my_index
  2. {
  3.   "settings": {
  4.     "number_of_shards": "2",
  5.     "number_of_replicas": "1",
  6.     "analysis": {
  7.       "analyzer": {
  8.         "ik_smart_pinyin": {
  9.           "type": "custom",
  10.           "tokenizer": "ik_smart",
  11.           "filter": [
  12.             "my_pinyin",
  13.             "word_delimiter"
  14.           ]
  15.         },
  16.         "ik_max_word_pinyin": {
  17.           "type": "custom",
  18.           "tokenizer": "ik_max_word",
  19.           "filter": [
  20.             "my_pinyin",
  21.             "word_delimiter"
  22.           ]
  23.         }
  24.       },
  25.       "filter": {
  26.         "my_pinyin": {
  27.           "type": "pinyin",
  28.           "keep_separate_first_letter": true,
  29.           "keep_full_pinyin": true,
  30.           "keep_original": true,
  31.           "limit_first_letter_length": 16,
  32.           "lowercase": true,
  33.           "remove_duplicated_term": true
  34.         }
  35.       }
  36.     }
  37.   },
  38.   "mappings": {
  39.     "_source": {
  40.       "enabled": true
  41.     },
  42.     "properties": {
  43.       "mmsi": {
  44.         "type": "text"
  45.       },
  46.       "name": {
  47.         "analyzer": "ik_max_word_pinyin",
  48.         "type": "text"
  49.       },
  50.       "pinyin": {
  51.         "type": "text"
  52.       }
  53.     }
  54.   }
  55. }

自定义模板

  1. ############################### template ##################
  2. {
  3.     "index_patterns": "index_*",
  4.     "order": 1000,
  5.     "settings": {
  6.         "number_of_shards": "2",
  7.         "number_of_replicas": "1"
  8.     },
  9.     "mappings": {
  10.         "_source": {
  11.             "enabled": true
  12.         },
  13.         "properties": {
  14.             "mmsi": {
  15.                 "type": "text"
  16.             },
  17.             "name": {
  18.                 "type": "text"
  19.             },
  20.             "k_name":{
  21.                 "type":"keyword"
  22.             },
  23.             "address":{
  24.                 "type":"text"
  25.             },
  26.             "p_name":{
  27.                 "type":"text"
  28.             }
  29.         }
  30.     },
  31.     "aliases": {}
  32. }
  33. ############################### template ##################
  35. input {
  36.   jdbc {
  37.     jdbc_driver_library => "D:/elk/logstash-7.11.1/test/postgresql-42.2.19.jar"
  38.     jdbc_driver_class => "org.postgresql.Driver"
  39.     jdbc_connection_string => "jdbc:postgresql://localhost:5432/waterway?characterEncoding=utf8&serverTimezone=GMT%2B8"
  40.     jdbc_user => "postgres"
  41.     jdbc_password => "953598751"
  42.     schedule => "*/20 * * * * *"
  43.     statement => "select id,st_astext(ST_Force2D(geom)) locations,name,pinyin p_name from add_anchor"
  44.     tracking_column => "id"
  45.     tracking_column_type => "numeric"
  46.     id => "add_padd_anchorort"
  47.     use_column_value => true
  48.     jdbc_default_timezone =>"Asia/Shanghai"
  49.     type => "add_anchor"
  50.   }
  51. }
  52. filter {
  53.   mutate {
  54.     add_field => { "k_name" => "%{name}" }
  55.   }
  56.   mutate {
  57.     remove_field => ["@version","@timestamp"]
  58.   }
  59. }
  60. output {
  61.     elasticsearch {
  62.       hosts => ["121.196.158.48:9200"]
  63.       index => "index_add_anchor"
  64.       document_id => "%{id}"
  65.       template =>"D:/elk/logstash-7.11.1/test/template/hd.json"
  66.       template_name => "index_my"
  67.       template_overwrite => true
  68.       # user => "elastic"
  69.       # password => "953598751"
  70.     }
  71. }

JavaUtil

  1. @Component
  2. public class EsUtil {
  3.     private static RestHighLevelClient client;
  5.     private EsUtil(@Autowired RestHighLevelClient restHighLevelClient){
  6.         client = restHighLevelClient;
  7.     }
  9.     /**
  10.      * 创建索引
  11.      * @param indexName     索引名称
  12.      */
  13.     public static boolean createIndex(String indexName) throws IOException {
  14.         GetIndexRequest getRequest = new GetIndexRequest(indexName);
  15.         if(client.indices().exists(getRequest, RequestOptions.DEFAULT)){
  16.             System.out.println("索引已经存在");
  17.             return false;
  18.         }
  19.         if(client.indices().create(new CreateIndexRequest(indexName),RequestOptions.DEFAULT).isAcknowledged()){
  20.             System.out.println("索引创建成功");
  21.             return true;
  22.         }
  23.         return false;
  24.     }
  26.     /**
  27.      * 判断所以是否已经存在
  28.      * @param indexName     索引名称
  29.      */
  30.     public static boolean indexExists(String indexName){
  31.         GetIndexRequest getRequest = new GetIndexRequest(indexName);
  32.         try {
  33.             if(client.indices().exists(getRequest, RequestOptions.DEFAULT)){
  34.                 System.out.println("索引已经存在");
  35.                 return false;
  36.             }
  37.         } catch (IOException e) {
  38.             e.printStackTrace();
  39.         }
  40.         return true;
  41.     }
  43.     /**
  44.      * 创建索引,对中文字段进行 ik+pinyin 分词
  45.      * @param indexName     索引名称
  46.      */
  47.     public static boolean createIndexIkPinyin(String indexName){
  48.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
  49.         // ik + pinyin 分词
  50.         createIndexRequest.settings("{\n" +
  51.                 "    \"number_of_shards\": \"2\",\n" +
  52.                 "    \"number_of_replicas\": \"1\",\n" +
  53.                 "    \"analysis\": {\n" +
  54.                 "      \"analyzer\": {\n" +
  55.                 "        \"ik_max_word_pinyin\": {\n" +
  56.                 "          \"type\": \"custom\",\n" +
  57.                 "          \"tokenizer\": \"ik_max_word\",\n" +
  58.                 "          \"filter\": [\n" +
  59.                 "            \"my_pinyin\",\n" +
  60.                 "            \"word_delimiter_graph\"\n" +
  61.                 "          ]\n" +
  62.                 "        }\n" +
  63.                 "      },\n" +
  64.                 "      \"filter\": {\n" +
  65.                 "        \"my_pinyin\": {\n" +
  66.                 "          \"type\": \"pinyin\",\n" +
  67.                 "          \"keep_separate_first_letter\": true,\n" +
  68.                 "          \"keep_full_pinyin\": true,\n" +
  69.                 "          \"keep_original\": true,\n" +
  70.                 "          \"limit_first_letter_length\": 16,\n" +
  71.                 "          \"lowercase\": true,\n" +
  72.                 "          \"remove_duplicated_term\": true\n" +
  73.                 "        }\n" +
  74.                 "      }\n" +
  75.                 "    }\n" +
  76.                 "  }",XContentType.JSON);
  77.         createIndexRequest.mapping("{\n" +
  78.                 "    \"_source\": {\n" +
  79.                 "      \"enabled\": true\n" +
  80.                 "    },\n" +
  81.                 "    \"properties\": {\n" +
  82.                 "      \"mmsi\": {\n" +
  83.                 "        \"type\": \"text\"\n" +
  84.                 "      },\n" +
  85.                 "      \"name\": {\n" +
  86.                 "        \"analyzer\": \"ik_max_word_pinyin\",\n" +
  87.                 "        \"type\": \"text\"\n" +
  88.                 "      },\n" +
  89.                 "      \"pinyin\": {\n" +
  90.                 "        \"type\": \"text\"\n" +
  91.                 "      }\n" +
  92.                 "    }\n" +
  93.                 "  }",XContentType.JSON);
  94.         CreateIndexResponse createIndexResponse = null;
  95.         try {
  96.             createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
  97.         } catch (IOException e) {
  98.             e.printStackTrace();
  99.         }
  100.         assert createIndexResponse != null;
  101.         return createIndexResponse.isAcknowledged();
  102.     }
  104.     /**
  105.      * 删除索引
  106.      * @param indexName     索引名称
  107.      */
  108.     public static boolean deleteIndex(String indexName){
  109.         DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
  110.         AcknowledgedResponse delete = null;
  111.         try {
  112.             delete = client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
  113.         } catch (IOException e) {
  114.             e.printStackTrace();
  115.         }
  116.         assert delete != null;
  117.         return delete.isAcknowledged();
  118.     }
  120.     /**
  121.      * 根据索引和 id 删除一条数据
  122.      * @param indexName     索引名称
  123.      * @param id            文档 id
  124.      */
  125.     public static boolean deleteIndexId(String indexName,String id){
  126.         DeleteRequest request = new DeleteRequest(indexName, id);
  127.         DeleteResponse delete1 = null;
  128.         try {
  129.             delete1 = client.delete(request, RequestOptions.DEFAULT);
  130.         } catch (IOException e) {
  131.             e.printStackTrace();
  132.         }
  133.         assert delete1 != null;
  134.         return delete1.getResult() == DocWriteResponse.Result.DELETED;
  135.     }
  137.     /**
  138.      * 根据索引和 id 删除一条数据
  139.      * @param indexName     索引
  140.      * @param ids           文档 id
  141.      */
  142.     public static boolean bulkDeleteIndexId(String indexName, Set<String> ids) {
  143.         BulkRequest bulkRequest = new BulkRequest();
  144.         for(String id : ids){
  145.             bulkRequest.add(new DeleteRequest(indexName).id(id));
  146.         }
  147.         BulkResponse bulk = null;
  148.         try {
  149.             bulk = client.bulk(bulkRequest, RequestOptions.DEFAULT);
  150.         } catch (IOException e) {
  151.             e.printStackTrace();
  152.         }
  153.         assert bulk != null;
  154.         return !bulk.hasFailures();
  155.     }
  157.     /**
  158.      * 插入数据
  159.      * @param indexName     索引
  160.      * @param id            文档 id
  161.      * @param data          数据,JSON格式
  162.      */
  163.     public static void insertData(String indexName, String id,String data) {
  164.         IndexRequest indexRequest = new IndexRequest(indexName);
  165.         indexRequest.id(id).source(data, XContentType.JSON);
  166.         try {
  167.             client.index(indexRequest, RequestOptions.DEFAULT);
  168.         } catch (IOException e) {
  169.             e.printStackTrace();
  170.         }
  171.     }
  173.     /**
  174.      * 批量插入数据
  175.      * @param indexName     索引
  176.      * @param data          数据(json)-- id,source
  177.      */
  178.     public static boolean bulkInsertData(String indexName, Map<String,String> data){
  179.         BulkRequest bulkRequest = new BulkRequest();
  180.         for(Map.Entry<String,String> entry : data.entrySet()){
  181.             bulkRequest.add(new IndexRequest(indexName).id(entry.getKey()).source(entry.getValue(),XContentType.JSON));
  182.         }
  183.         BulkResponse bulk = null;
  184.         try {
  185.             bulk = client.bulk(bulkRequest, RequestOptions.DEFAULT);
  186.         } catch (IOException e) {
  187.             e.printStackTrace();
  188.         }
  189.         assert bulk != null;
  190.         return !bulk.hasFailures();
  191.     }
  193.     /**
  194.      * 查询数据
  195.      * https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.x/java-rest-high-search.html
  196.      * @param indexName     索引
  197.      */
  198.     public static Object search(String indexName) throws IOException {
  199.         SearchRequest searchRequest = new SearchRequest(indexName);
  200.         // 查询条件
  201.         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
  202.         searchSourceBuilder.query(QueryBuilders.matchAllQuery());
  203.         searchSourceBuilder.from(0);
  204.         searchSourceBuilder.size(10);
  205.         searchSourceBuilder.timeout(new TimeValue(3, TimeUnit.SECONDS));
  206. //        // 按 _score 降序排序(默认)
  207. //        searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));
  208. //        // 按 _id 字段升序排序
  209. //        searchSourceBuilder.sort(new FieldSortBuilder("id").order(SortOrder.ASC));
  211.         // 设置返回结果中包含或排除那些字段
  212. //        String[] includeFields = new String[] {"title", "innerObject.*"};
  213. //        String[] excludeFields = new String[] {"user"};
  214. //        searchSourceBuilder.fetchSource(includeFields, excludeFields);
  216.         searchRequest.source(searchSourceBuilder);
  217.         SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
  218.         SearchHits hits = search.getHits();
  220.         TotalHits totalHits = hits.getTotalHits();
  221.         System.out.println(totalHits.value);
  223.         System.out.println(hits.getMaxScore());
  225.         SortField[] sortFields = hits.getSortFields();
  227.         SearchHit[] hits1 = hits.getHits();
  228.         for(SearchHit hit : hits1){
  229.             System.out.println(hit.getIndex());
  230.             System.out.println(hit.getId());
  231.         }
  232.         return "";
  233.     }
  234. }