filebeat6.1.3版本的pipeline配置问题，请大神指导，谢谢！

ES：6.1.3
filebeat:6.1.3
目的：提取日志内容的“时间”，“级别”，组成新的fields
 
就是把下面的日志内容：
日志例子："2019-08-16 20:48:57,459 | INFO  | DataNode: [[[DISK]file:/srv/BigData/hadoop/data1/dn/, [DISK]file:/srv/BigData/hadoop/data2/dn/]] heartbeating to mndn03/10.7.9.13:25006 | Scheduling blk_1091327814_17598925 file /srv/BigData/hadoop/data1/dn/current/BP-1637367359-10.7.9.13-1546141567141/current/finalized/subdir12/subdir87/blk_1091327814 for deletion | FsDatasetAsyncDiskService.java:216",
                    "source": "/var/log/Bigdata/hdfs/dn/hadoop-omm-datanode-mndn01.log"
 
时间，级别 提取出来，通过pipeline完成提取，最后写入index
 
 
ES pipeline配置：
PUT _ingest/pipeline/sxapp_access
{
    "description" : "sxapp access log pipeline",
    "processors": [
        {
            "grok": {
                "field": "message",
                "patterns": [
                    "%{TIMESTAMP_ISO8601:timestamp} | %{DATA:level} | %{DATA:content}"
                ]
            }
        }
    ]
}
 
ES索引配置：
{
    "settings": {
        "number_of_shards":3,
        "number_of_replicas":1,
        "index": {
            "analysis.analyzer,default.type": "ik_smart"
        }
    },
    "mappings": {
        "doc": {
            "_source": {
                "enabled": true
            },
            "properties": {
                "server_ip": {
                    "type": "text"
                },
                "server_name": {
                    "type": "text"
                },
                "message": {
                    "type": "text"
                }
            }
        }
    }
}
 
filebeat配置：
filebeat.prospectors:

- type: log

  enabled: true

  paths:
    - /var/log/Bigdata/hdfs/dn/hadoop-omm-datanode-mndn01.log

  fields:
    type: sxapp_access


filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.settings:
  index.number_of_shards: 3

setup.kibana:

output.elasticsearch:
  hosts: ['10.7.9.11:24100','10.7.9.12:24100','10.7.9.13:24100','10.7.9.14:24100']
  pipelines:
    - pipeline: sxapp_access
      when.equals:
        fields.type: sxapp_access

  index: my_index_01
setup.template.name: my_index_01
setup.template.pattern: my_index_01
 
 
启动报错：
2019/08/17 07:18:50.965399 client.go:276: ERR Failed to perform any bulk index operations: 500 Internal Server Error: {"error":{"root_cause":[{"type":"illegal_state_exception","reason":"There are no ingest nodes in this cluster, unable to forward request to an ingest node."}],"type":"illegal_state_exception","reason":"There are no ingest nodes in this cluster, unable to forward request to an ingest node."},"status":500}
2019/08/17 07:18:50.971247 client.go:276: ERR Failed to perform any bulk index operations: 500 Internal Server Error: {"error":{"root_cause":[{"type":"illegal_state_exception","reason":"There are no ingest nodes in this cluster, unable to forward request to an ingest node."}],"type":"illegal_state_exception","reason":"There are no ingest nodes in this cluster, unable to forward request to an ingest node."},"status":500}