参考：官方链接 参数说明： processors: 内置pipeline支持数据加工处理器 set：使用集合新增数据

创建第一个pipeline

案例：增加省市区

  1. GET _ingest/pipeline/mypipline-001
  3. PUT _ingest/pipeline/mypipline-001
  4. {
  5.   "description": "增加省市区数据",
  6.   "processors": [
  7.     {
  8.       "set": {
  9.         "field": "pro",
  10.         "value": "HN"
  11.       }
  12.     }
  13.   ]
  14. }
  16. POST ckiss-company-003/_doc?pipeline=mypipline-001
  17. {
  18.   "companyName": "ckiss Company"
  19. }
  21. GET ckiss-company-003/_mapping
  1. {
  2. "ckiss-company-003" : {
  3.   "mappings" : {
  4.     "properties" : {
  5.       "companyName" : {
  6.         "type" : "text",
  7.         "fields" : {
  8.           "keyword" : {
  9.             "type" : "keyword",
  10.             "ignore_above" : 256
  11.           }
  12.         }
  13.       },
  14.       "pro" : {
  15.         "type" : "text",
  16.         "fields" : {
  17.           "keyword" : {
  18.             "type" : "keyword",
  19.             "ignore_above" : 256
  20.           }
  21.         }
  22.       }
  23.     }
  24.   }
  25. }
  26. }

支持对象类型

  1. PUT _ingest/pipeline/mypipline-002
  2. {
  3.   "description": "增加省市区数据",
  4.   "processors": [
  5.     {
  6.       "set": {
  7.         "field": "area.pro",
  8.         "value": "HN"
  9.       }
  10.     },
  11.     {
  12.       "set": {
  13.         "field": "area.city",
  14.         "value": "CS"
  15.       }
  16.     },
  17.     {
  18.       "set": {
  19.         "field": "area.country",
  20.         "value": "CN"
  21.       }
  22.     }
  23.   ]
  24. }
  26. POST ckiss-company-004/_doc?pipeline=mypipline-002
  27. {
  28.   "companyName": "ckiss Company"
  29. }
  31. GET ckiss-company-004/_mapping
  33. GET  ckiss-company-004/_search

响应

  1. "_source" : {
  2.   "area" : {
  3.     "country" : "CN",
  4.     "city" : "CS",
  5.     "pro" : "HN"
  6.   },
  7.   "companyName" : "ckiss Company"
  8. }

经过管道的话，管道有最终决定权

数据格式不规范

  1. POST ckiss-company-003/_doc?pipeline=mypipline-002
  2. {
  3.   "companyName": "ckiss国际集团",
  4.   "area.pro": "JS",
  5.   "area.city": "SZ"
  6. }

响应
ingest错误数据.png
正确格式

  1. POST ckiss-company-003/_doc?pipeline=mypipline-002
  2. {
  3.   "companyName": "ckiss国际集团",
  4.   "area": {
  5.     "pro": "JS",
  6.     "city": "SZ"
  7.   }
  8. }

_simulate模拟测试

  1. POST _ingest/pipeline/mypipline-002/_simulate
  2. {
  3.   "docs": [
  4.     {
  5.       "_source": {
  6.         "area": {
  7.           "pro": "SH"
  8.         }
  9.       }
  10.     }
  11.   ]
  12. }
  14. POST _ingest/pipeline/mypipline-002/_simulate
  15. {
  16.   "docs": [
  17.     {
  18.       "_source": {}
  19.     }
  20.   ]
  21. }
  23. # 以上结果一样

访问_source源数据

  1. DELETE _ingest/pipeline/mypipline-002
  3. PUT _ingest/pipeline/mypipline-002
  4. {
  5.   "description": "source源数据上下文",
  6.   "processors": [
  7.     {
  8.       "set": {
  9.         "field": "company.companyId",
  10.         "value": "{{_source.companyId}}"
  11.       }
  12.     },
  13.     {
  14.       "set": {
  15.         "field": "company.companyName",
  16.         "value": "{{_source.companyName}}"
  17.       }
  18.     },
  19.     {
  20.       "remove": {
  21.         "field": [
  22.           "companyId",
  23.           "companyName"
  24.         ]
  25.       }
  26.     }
  27.   ]
  28. }
  30. DELETE ckiss-company-001
  32. PUT ckiss-company-001/_doc/1?pipeline=mypipline-002
  33. {
  34.   "companyId":1,
  35.   "companyName": "ckiss国际集团",
  36.   "city": "WH"
  37. }
  39. GET ckiss-company-001/_search

访问meta元数据

  • 元数据上下文参数：index、type、id、routing
  • 注意：_id如果是自动生成，则无法获取
  • 访问的语法通过: {{_index}} ```json DELETE _ingest/pipeline/mypipline-002

PUT _ingest/pipeline/mypipline-002 { “description”: “meta元数据_id”, “processors”: [ { “set”: { “field”: “company.companyId”, “value”: “{{_id}}” } }, { “remove”: { “field”: [ “companyId”, “companyName” ] } } ] }

GET _ingest/pipeline/mypipline-002

DELETE ckiss-company-001

PUT ckiss-company-001/_doc/122?pipeline=mypipline-002 { “companyId”:1233, “companyName”: “ckiss国际集团”, “city”: “WH” }

GET ckiss-company-001/_search

  1. **响应**
  2. ```json
  3. {
  4.   "_index" : "ckiss-company-001",
  5.   "_type" : "_doc",
  6.   "_id" : "122",
  7.   "_score" : 1.0,
  8.   "_source" : {
  9.     "city" : "WH",
  10.     "company" : {
  11.       "companyId" : "122"
  12.     }
  13.   }
  14. }

访问ingest元数据

  • ingest元数据参数： _ingest.timestamp
  • 访问语法：{{_ingest.timestamp}} ```json DELETE _ingest/pipeline/mypipline-002

PUT _ingest/pipeline/mypipline-002 { “processors”: [ { “set”: { “field”: “@timestamp”, “value”: “{{_ingest.timestamp}}” } } ] }

  2. <a name="mcYAI"></a>
  3. ### 多管道
  4. ```json
  5. PUT _ingest/pipeline/company-area-001
  6. {
  7.   "description": "区域讯息", 
  8.    "processors": [
  9.     {
  10.       "set": {
  11.         "field": "area.pro",
  12.         "value": "HN"
  13.       }
  14.     },
  15.     {
  16.       "set": {
  17.         "field": "area.city",
  18.         "value": "CS"
  19.       }
  20.     },
  21.     {
  22.       "set": {
  23.         "field": "area.country",
  24.         "value": "CN"
  25.       }
  26.     }
  27.   ]
  28. }
  30. PUT  _ingest/pipeline/company-info-001
  31. {
  32.    "description": "公司基本讯息", 
  33.    "processors": [
  34.      {
  35.        "set": {
  36.          "field": "companyName",
  37.          "value": "国际集团"
  38.        }
  39.      }
  40.    ]
  41. }
  43. PUT _ingest/pipeline/company-remove-001
  44. {
  45.   "description": "删除字段",
  46.   "processors": [
  47.     {
  48.       "set": {
  49.         "field": "@timestamp",
  50.         "value": "{{_ingest.timestamp}}"
  51.       }
  52.     },
  53.     {
  54.       "remove": {
  55.         "field": [
  56.           "companyId"
  57.           ]
  58.       }
  59.     }
  60.   ]
  61. }
  63. PUT _ingest/pipeline/company_pipeline_001
  64. {
  65.   "description": "多管道处理器",
  66.   "processors": [
  67.     {
  68.       "pipeline": {
  69.         "name": "company-area-001"
  70.       }
  71.     },
  72.     {
  73.       "pipeline": {
  74.         "name": "company-info-001"
  75.       }
  76.     },
  77.     {
  78.       "pipeline": {
  79.         "name": "/company-remove-001"
  80.       }
  81.     }
  82.   ]
  83. }

script脚本处理

  1. PUT _ingest/pipeline/company_script_001
  2. {
  3.   "description": "脚本处理器",
  4.   "processors": [
  5.     {
  6.       "script": {
  7.         "source": """
  8.         ctx.companyNameId=ctx.companyName+"_"+ctx.companyId;
  9.         """,
  10.         "lang": "painless",
  11.         "params": {}
  12.       }
  13.     }
  14.   ]
  15. }
  17. GET _ingest/pipeline/company_script_001
  19. DELETE ckiss-company-001
  21. PUT ckiss-company-001/_doc/12?pipeline=company_script_001
  22. {
  23.   "companyId":2,
  24.   "userId": 123,
  25.   "userName": "测试数据",
  26.   "regDate": "2022-04-10",
  27.   "companyName": "ckiss国际集团"
  28. }
  30. GET ckiss-company-001/_search

if逻辑判断

  • 几乎所有的处理器都支持逻辑表达式，if 里面编写的是脚本

    根据公司ID判断公司基本信息

  1. DELETE _ingest/pipeline/company_info_003
  2. PUT _ingest/pipeline/company_info_003
  3. {
  4.   "description": "公司信息重新组织",
  5.   "processors": [
  6.     {
  7.       "set": {
  8.         "if": "ctx._id=='1'",
  9.         "field": "companyId",
  10.         "value": "{{_id}}"
  11.       }
  12.     },
  13.     {
  14.       "set": {
  15.         "if": "ctx.companyId=='5'",
  16.         "field": "companyName",
  17.         "value": "{{_source.userName}}"
  18.       }
  19.     }
  20.   ]
  21. }
  23. DELETE ckiss-company-001
  25. PUT ckiss-company-001/_doc/5?pipeline=company_info_003
  26. {
  27.   "companyId": 5,
  28.   "city": "wh",
  29.   "userId": 123,
  30.   "userName": "测试数据",
  31.   "regDate": "2022-04-10",
  32.   "companyName": "ckiss国际集团"
  33. }
  34. GET ckiss-company-001/_search

索引重建pipeline

  1. POST _reindex
  2. {
  3.   "conflicts": "proceed",
  4.   "source": {
  5.     "index": "source-index"
  6.   },
  7.   "dest": {
  8.     "index": "desc-index",
  9.     "pipeline": "pipeline"
  10.   }
  11. }