Skip to content
Generate Elasticsearch indexes based on Table Schema descriptors.
Python Makefile
Branch: master
Clone or download
Latest commit 7c06a18 Dec 18, 2019
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.github
data
examples Sync readme with other tableschema plugins (#10) Sep 28, 2017
tableschema_elasticsearch
tests
.env.example Initial commit Jul 24, 2017
.gitignore
.travis.yml
LEAD.md
LICENSE.md
MANIFEST.in
Makefile
README.md
pylama.ini
pytest.ini
setup.cfg
setup.py
tox.ini

README.md

tableschema-elasticsearch-py

Travis Coveralls PyPi Github Gitter

Generate and load ElasticSearch indexes based on Table Schema descriptors.

Features

  • implements tableschema.Storage interface

Contents

Getting Started

Installation

The package use semantic versioning. It means that major versions could include breaking changes. It's highly recommended to specify package version range in your setup/requirements file e.g. package>=1.0,<2.0.

pip install tableschema-elasticsearch

Documentation

Usage overview

import elasticsearch
import jsontableschema_es

INDEX_NAME = 'testing_index'

# Connect to Elasticsearch instance running on localhost
es=elasticsearch.Elasticsearch()
storage=jsontableschema_es.Storage(es)

# List all indexes
print(list(storage.buckets))

# Create a new index
storage.create('test', [
    ('numbers',
     {
         'fields': [
             {
                 'name': 'num',
                 'type': 'number'
             }
         ]
     })
])

# Write data to index
l=list(storage.write(INDEX_NAME, 'numbers', ({'num':i} for i in range(1000)), ['num']))
print(len(l))
print(l[:10], '...')

l=list(storage.write(INDEX_NAME, 'numbers', ({'num':i} for i in range(500,1500)), ['num']))
print(len(l))
print(l[:10], '...')

# Read all data from index
storage=jsontableschema_es.Storage(es)
print(list(storage.buckets))
l=list(storage.read(INDEX_NAME))
print(len(l))
print(l[:10])

In this driver elasticsearch is used as the db wrapper. We can get storage this way:

from elasticsearch import Elasticsearch
from jsontableschema_sql import Storage

engine = Elasticsearch()
storage = Storage(engine)

Then we could interact with storage ('buckets' are ElasticSearch indexes in this context):

storage.buckets # iterator over bucket names
storage.create('bucket', [(doc_type, descriptor)],
               reindex=False,
               always_recreate=False,
               mapping_generator_cls=None)
        # doc_type can be None in case mapping_types are not supported (ES version >= 7.0.0)
        # reindex will copy existing documents from an existing index with the same name (in case of a mapping conflict)
        # always_recreate will always recreate an index, even if it already exists. default is to update mappings only.
        # mapping_generator_cls allows customization of the generated mapping
storage.delete('bucket')
storage.describe('bucket') # return descriptor, not implemented yet
storage.iter('bucket', doc_type=optional) # yield rows
storage.read('bucket', doc_type=optional) # return rows
storage.write('bucket', doc_type, rows, primary_key,
              as_generator=False)
        # primary_key is a list of field names which will be used to generate document ids

When creating indexes, we always create an index with a semi-random name and a matching alias that points to it. This allows us to decide whether to re-index documents whenever we're re-creating an index, or to discard the existing records.

Mappings

When creating indexes, the tableschema types are converted to ES types and a mapping is generated for the index.

Some special properties in the schema provide extra information for generating the mapping:

  • array types need also to have the es:itemType property which specifies the inner data type of array items.
  • object types need also to have the es:schema property which provides a tableschema for the inner document contained in that object (or have es:enabled=false to disable indexing of that field).

Example:

{
  "fields": [
    {
      "name": "my-number",
      "type": "number"
    },
    {
      "name": "my-array-of-dates",
      "type": "array",
      "es:itemType": "date"
    },
    {
      "name": "my-person-object",
      "type": "object",
      "es:schema": {
        "fields": [
          {"name": "name", "type": "string"},
          {"name": "surname", "type": "string"},
          {"name": "age", "type": "integer"},
          {"name": "date-of-birth", "type": "date", "format": "%Y-%m-%d"}
        ]
      }
    },
    {
      "name": "my-library",
      "type": "array",
      "es:itemType": "object",
      "es:schema": {
        "fields": [
          {"name": "title", "type": "string"},
          {"name": "isbn", "type": "string"},
          {"name": "num-of-pages", "type": "integer"}
        ]
      }
    },
    {
      "name": "my-user-provded-object",
      "type": "object",
      "es:enabled": false
    }
  ]
}

Custom mappings

By providing a custom mapping generator class (via mapping_generator_cls), inheriting from the MappingGenerator class you should be able

API Reference

Storage

Storage(self, es=None)

Elasticsearch Tabular Storage.

Package implements Tabular Storage interface (see full documentation on the link):

Storage

Only additional API is documented

Arguments

  • es (object): ElasticSearch instance

storage.create

storage.create(self, bucket, doc_types, reindex=False, always_recreate=False, mapping_generator_cls=None, index_settings=None)

Create index with mapping by schema.

Arguments

  • bucket(str): Name of index to be created
  • doc_types(list<(doc_type, descriptor)>): List of tuples of doc_types and matching descriptors
  • always_recreate: Delete index if already exists (otherwise just update mapping)
  • reindex: On mapping mismath, automatically create new index and migrate existing indexes to it
  • mapping_generator_cls: subclass of MappingGenerator
  • index_settings: settings which will be used in index creation

storage.delete

storage.delete(self, bucket=None)

Delete index with mapping by schema.

Arguments

  • bucket(str): Name of index to delete

Contributing

The project follows the Open Knowledge International coding standards.

Recommended way to get started is to create and activate a project virtual environment. To install package and development dependencies into active environment:

$ make install

To run tests with linting and coverage:

$ make test

Changelog

Here described only breaking and the most important changes. The full changelog and documentation for all released versions could be found in nicely formatted commit history.

v1.0

  • Initial driver implementation
You can’t perform that action at this time.