\n\nWhat is Pelias?\n \nPelias is a search engine for places worldwide, powered by open data. It turns addresses and place names into geographic coordinates, and turns geographic coordinates into places and addresses. With Pelias, you're able to turn your users' place searches into actionable geodata and transform your geodata into real places.\n
\nWe think open data, open source, and open strategy win over proprietary solutions at any part of the stack and we want to ensure the services we offer are in line with that vision. We believe that an open geocoder improves over the long-term only if the community can incorporate truly representative local knowledge.\n\n\n# Pelias Elasticsearch database client\n\nThis module provides a Node.js stream for bulk-inserting documents into [Elasticsearch](https://www.elastic.co/products/elasticsearch).\n\n\n## Install Dependencies\n\n\n```bash\n$ npm install\n```\n\n## Usage\nThis module returns “streamFactory” —a function that produces a transforming stream. The stream puts documents into elasticsearch during import pipeline. Note: this stream triggers finish event only after all documents are stored into elasticsearch.\n\n\n```javascript\n'use strict';\n\n// some_importer.js\n\nconst streamify = require('stream-array');\nconst through = require('through2');\nconst Document = require('pelias-model').Document;\nconst dbMapper = require('pelias-model').createDocumentMapperStream;\nconst dbclient = require('pelias-dbclient');\n\nconst elasticsearch = require('elasticsearch');\nconst config = require('pelias-config').generate();\nconst elasticDeleteQuery = require('elastic-deletebyquery');\n\nconst timestamp = Date.now();\n\nconst stream = streamify([1, 2, 3])\n .pipe(through.obj((item, enc, next) => {\n const uniqueId = [ 'docType', item ].join(':'); // documents with the same id will be updated\n const doc = new Document( 'sourceType', 'venue', uniqueId );\n doc.timestamp = timestamp;\n next(null, doc);\n }))\n .pipe(dbMapper())\n .pipe(dbclient()); // put documents into elasticsearch\n\nstream.on('finish', () => {\n const client = new elasticsearch.Client(config.esclient);\n elasticDeleteQuery(client);\n\n const options = {\n index: config.schema.indexName,\n body: {\n query: {\n \"bool\": {\n \"must\": [\n {\"term\": { \"source\": \"sourceType\" }}\n ],\n \"must_not\": [\n {\"term\": { \"timestamp\": timestamp }}\n ]\n }\n }\n }\n };\n\n client.deleteByQuery(options, (err, response) => {\n console.log('The elements deleted are: %s', response.elements);\n });\n});\n\n```\n\n## Configuration\n\nSettings are read from the `dbclient` section of your `pelias.json`:\n\n```javascript\n{\n \"dbclient\": {\n \"statFrequency\": 10000,\n \"batchSize\": 500,\n \"flooding\": {\n \"pause\": 5,\n \"resume\": 2\n }\n }\n}\n```\n\n| setting | default | meaning |\n|---|---|---|\n| `statFrequency` | 10000 | how often, in milliseconds, progress is logged |\n| `batchSize` | 500 | documents per bulk request |\n| `flooding.pause` | 5 | bulk requests allowed in flight before the stream is paused |\n| `flooding.resume` | 2 | in-flight requests the stream drains to before resuming |\n\nAny of these can also be passed directly to the stream factory, which takes\npriority over the configuration file.\n\n## Contributing\n\nPlease fork and pull request against upstream master on a feature branch.\n\nPretty please; provide unit tests and script fixtures in the `test` directory.\n\n### Running Unit Tests\n\n```bash\n$ npm test\n```\n\n### Continuous Integration\n\nCI tests every release against all currently supported Node.js versions.\n","readmeFilename":"README.md"}
![](\"https://raw.githubusercontent.com/pelias/design/master/logo/pelias_github/Github_markdown_hero.png\"){kind=logo}
\n![](\"https://github.com/pelias/dbclient/workflows/Continuous%20Integration/badge.svg\"){kind=icon}
\n![](\"https://img.shields.io/github/license/pelias/dbclient?style=flat&color=orange\")
![](\"https://img.shields.io/gitter/room/pelias/pelias?style=flat&color=yellow\")