Computer >> Hướng Dẫn Máy Tính >  >> Lập Trình >> Cơ Sở Dữ Liệu

Parse CSV Files liền mạch với Filebeat và Elasticsearch Ingest Pipelines

Một trong coolest new features Elasticsearch 5 là ingest node, thêm Logstash-style processing vào Elasticsearch cluster, transform data trước khi index without needing another service/infrastructure. Trước đó, chúng tôi post quick blog parse csv files với Logstash, nên tôi provide ingest pipeline version để compare.

Đây show example dùng Filebeat ship data đến ingest pipeline, index, visualize với Kibana.

The Data

Nhiều great sources free data, nhưng vì most ObjectRocket team ở Austin, TX, dùng data từ data.austintexas.gov. Restaurant inspection data set good size với enough relevant info cho real world example.

Vài lines từ dataset cho idea structure:

Restaurant Name,Zip Code,Inspection Date,Score,Address,Facility ID,Process Description
Westminster Manor,78731,07/21/2015,96,"4100 JACKSON AVE AUSTIN, TX 78731 (30.314499, -97.755166)",2800365,Routine Inspection
Wieland Elementary,78660,10/02/2014,100,"900 TUDOR HOUSE RD AUSTIN, TX 78660 (30.422862, -97.640183)",10051637,Routine Inspection

DOH… Không phải nice single line per entry case, nhưng OK. Sắp thấy tools Elastic Stack gives work around scenarios.

Setting up Filebeat

First step: get Filebeat ready ship data to Elasticsearch cluster. Sau khi download Filebeat (same version ES cluster) và extract, extremely simple setup via included filebeat.yml. Scenario này, config tôi dùng:

filebeat.prospectors:
- input_type: log
  paths:
    - /Path/To/logs/*.csv
  # Ignore first line with column headings
  exclude_lines: ["^Restaurant Name,"]
  # Identifies last two columns as end of entry then prepends previous lines to it
  multiline.pattern: ',\d+,[^\",]+$'
  multiline.negate: true
  multiline.match: before
#================================ Outputs =====================================
output.elasticsearch:
  hosts: ["https://dfw-xxxxx-0.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-1.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-2.es.objectrocket.com:xxxxx", "https://dfw-xxxxx-3.es.objectrocket.com:xxxxx"]
  pipeline: "inspectioncsvs"
  username: "esuser"
  password: "supersecretpassword"

Everything straightforward: section specify where/how grab input files, section specify where ship data. Chỉ call out multiline bit và Elasticsearch config piece.

Vì formatting dataset không super strict, inconsistent double quotes và newlines sprinkled in, best option: look for end of entry - numeric ID followed by inspection type without variation/double-quotes/newlines. Từ đó, Filebeat queue up unmatched lines và prepend to final line matching pattern. Nếu data cleaner, simple line per entry format, pretty much ignore multiline settings.

Elasticsearch output section: standard settings với small addition pipeline name với pipeline: directive. Nếu ObjectRocket service, grab output snippet từ "Connect" tab UI, prepopulated right hosts, thêm pipeline line fill user/password. Ensure system IP added to cluster ACLs.

Creating the Ingest Pipeline

Input data và Filebeat ready, create và tweak ingest pipeline. Main tasks pipeline cần perform:

  • Split csv content thành correct fields
  • Convert inspection score thành integer
  • Set @timestamp field
  • Clean up other data formatting

Pipeline làm all that:

PUT _ingest/pipeline/inspectioncsvs
{
  "description" : "Convert Restaurant inspections csv data to indexed data",
  "processors" : [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{REST_NAME:RestaurantName},%{REST_ZIP:ZipCode},%{MONTHNUM2:InspectionMonth}/%{MONTHDAY:InspectionDay}/%{YEAR:InspectionYear},%{NUMBER:Score},\"%{DATA:StreetAddress}\n%{DATA:City},?\\s+%{WORD:State}\\s*%{NUMBER:ZipCode2}\\s*\n\\(?%{DATA:Location}\\)?\",%{NUMBER:FacilityID},%{DATA:InspectionType}$"],
        "pattern_definitions": {
          "REST_NAME": "%{DATA}|%{QUOTEDSTRING}",
          "REST_ZIP": "%{QUOTEDSTRING}|%{NUMBER}"
        }
      }
    },
    {
      "grok": {
        "field": "ZipCode",
        "patterns": [".*%{ZIP:ZipCode}\"?$"],
        "pattern_definitions": {
          "ZIP": "\\d{5}"
        }
      }
    },
    {
      "convert": {
        "field" : "Score",
        "type": "integer"
      }
    },
    {
      "set": {
        "field" : "@timestamp",
        "value" : "//"
      }
    },
    {
      "date" : {
        "field" : "@timestamp",
        "formats" : ["yyyy/MM/dd"]
      }
    }
  ],
  "on_failure" : [
    {
      "set" : {
        "field" : "error",
        "value" : " - Error processing message - "
      }
    }
  ]
}

Khác Logstash, ingest pipeline không có csv processor/plugin (time of writing), nên transform csv's yourself. Dùng grok processor heavy lifting, vì each row chỉ few columns. Data nhiều columns hơn, grok processor pretty hairy, option khác: split processor và painless scripting process line iterative fashion. Second grok processor deal với two ways zip codes entered trong dataset.

Debug purposes, include general on_failure section catch all errors print processor type failed và message broke pipeline. Makes debug way easier. Query index documents có error set, debug với simulate API.

Testing the Pipeline

Ingest pipeline configured, test và run với simulate API. First cần sample document. Couple ways: run Filebeat without pipeline setting grab unprocessed document từ Elasticsearch, hoặc run Filebeat với console output enabled, comment out Elasticsearch section thêm:

output.console:
  pretty: true

Sample document từ environment:

POST _ingest/pipeline/inspectioncsvs/_simulate
{
  "docs" : [
    {
      "_index": "inspections",
      "_type": "log",
      "_id": "AVpsUYR_du9kwoEnKsSA",
      "_score": 1,
      "_source": {
        "@timestamp": "2017-03-31T18:22:25.981Z",
        "beat": {
          "hostname": "systemx",
          "name": "RestReviews",
          "version": "5.1.1"
        },
        "input_type": "log",
        "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection",
        "offset": 2109798,
        "source": "/Path/to/my/logs/Restaurant_Inspection_Scores.csv",
        "tags": [ "debug", "reviews" ],
        "type": "log"
      }
    }
  ]
}

Response (trimmed to fields we tried to set):

{
  "docs": [
    {
      "doc": {
        "_id": "AVpsUYR_du9kwoEnKsSA",
        "_type": "log",
        "_index": "inspections",
        "_source": {
          "InspectionType": "Routine Inspection",
          "ZipCode": "78660",
          "InspectionMonth": "10",
          "City": "AUSTIN",
          "message": "Wieland Elementary,78660,10/02/2014,100,\"900 TUDOR HOUSE RD\nAUSTIN, TX 78660\n(30.422862, -97.640183)\",10051637,Routine Inspection",
          "RestaurantName": "Wieland Elementary",
          "FacilityID": "10051637",
          "Score": 100,
          "StreetAddress": "900 TUDOR HOUSE RD",
          "State": "TX",
          "InspectionDay": "02",
          "InspectionYear": "2014",
          "ZipCode2": "78660",
          "Location": "30.422862, -97.640183"
        },
        "_ingest": {
          "timestamp": "2017-03-31T20:36:59.574+0000"
        }
      }
    }
  ]
}

Pipeline definitely succeeded, all data appears right place.

Running filebeat

Trước khi run Filebeat, last thing. Optional nếu just want comfortable với ingest pipeline, nhưng nếu muốn dùng Location field set trong grok processor as Geo-point, cần add mapping to filebeat.template.json file, thêm vào properties section:

"Location": {
  "type": "geo_point"
},

Fire up Filebeat: ./filebeat -e -c filebeat.yml -d "elasticsearch".

Using the data

GET /filebeat-*/_count
{}
{
  "count": 25081,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  }
}

Good sign! Check errors:

GET /filebeat-*/_search
{
  "query": {
    "exists" : {
      "field" : "error"
    }
  }
}
{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  },
  "hits": {
    "total": 0,
    "max_score": null,
    "hits": []
  }
}

Another good sign!

Ready visualize và show data in Kibana. Walk through creating Kibana dashboards another time, nhưng với dates, restaurant names, scores, locations, có plenty freedom tạo cool visualizations.

Final notes

Ingest pipeline pretty powerful handle transformations easily. Move all processing to Elasticsearch, chỉ dùng lightweight Beats trên hosts, without requiring Logstash. Tuy nhiên, vẫn gaps ingest node vs Logstash. Ví dụ: number of processors available in ingest pipeline still limited, simple tasks như parsing CSV không easy như Logstash. Elasticsearch team regularly rolling out new processors, hoping list of differences keeps getting smaller.