Browse documentation

#ETL Adapter: Seal

This is the Flow PHP Adapter for SEAL. SEAL is a PHP Search Engine Abstraction Layer with one API for many search engines (Algolia, Elasticsearch, Meilisearch, OpenSearch, Solr, Typesense, RediSearch, Loupe). With this adapter, your Flow PHP ETL pipelines can load data into each search backend that SEAL supports. The pipelines do not depend on a specific search engine implementation.

#Installation

For the installation instructions, read the installation page.

Also install the SEAL engine adapter for your search backend, for example:

composer require flow-php/etl-adapter-seal cmsig/seal-elasticsearch-adapter

#Description

The adapter works on a CmsIg\Seal\EngineInterface. You build it yourself from the SEAL adapter of your choice and a SEAL schema. The Doctrine and PostgreSQL adapters work in the same way: they accept a pre-configured connection/client. It also works with the SEAL Symfony bundle, which can inject a fully configured engine.

• Loader: to_seal_upsert() / to_seal_delete() write or remove documents in a search index in bulks.
• Schema Conversion: to_seal_schema() / seal_schema_to_flow() convert between Flow and SEAL schemas.

#Building an Engine

use CmsIg\Seal\Adapter\Elasticsearch\ElasticsearchAdapter;
use CmsIg\Seal\Engine;

use function Flow\ETL\Adapter\Seal\to_seal_schema;
use function Flow\ETL\DSL\{int_schema, schema, str_schema};

$sealSchema = to_seal_schema(
    schema(
        str_schema('id'),
        str_schema('name'),
        int_schema('age'),
    ),
    index_name: 'users',
    identifier: 'id',
);

$engine = new Engine(new ElasticsearchAdapter($client), $sealSchema);
$engine->createIndex('users');

#Loader

#Upserting Documents

to_seal_upsert() saves documents into the index. Search engines upsert by document identifier. A document with an existing identifier replaces the old document.

use function Flow\ETL\Adapter\Seal\to_seal_upsert;
use function Flow\ETL\Adapter\CSV\from_csv;
use function Flow\ETL\DSL\data_frame;

data_frame()
    ->read(from_csv(__DIR__ . '/users.csv'))
    ->write(to_seal_upsert($engine, 'users'))
    ->run();

#Bulk Size

By default, documents are written in bulks of 100. To change this, use withBulkSize():

to_seal_upsert($engine, 'users')->withBulkSize(500);

#Deleting Documents

to_seal_delete() removes documents by their identifiers, which it reads from the id column. withIdentifierColumn() names a different column.

use function Flow\ETL\Adapter\Seal\to_seal_delete;
use function Flow\ETL\DSL\{data_frame, from_array};

data_frame()
    ->read(from_array([['id' => '1'], ['id' => '2']]))
    ->write(to_seal_delete($engine, 'users'))
    ->run();

to_seal_delete($engine, 'products')->withIdentifierColumn('sku');

#Data Normalization

Each row becomes one document:

#Schema Conversion

to_seal_schema() converts a Flow schema into a SEAL schema for one index. SEAL needs exactly one identifier field for each index. Name it with the identifier argument, or mark a definition with SealMetadata::identifier().

use Flow\ETL\Adapter\Seal\Values\Schema\SealMetadata;

use function Flow\ETL\Adapter\Seal\to_seal_schema;
use function Flow\ETL\DSL\{datetime_schema, float_schema, int_schema, schema, str_schema};

$sealSchema = to_seal_schema(
    schema(
        str_schema('id'),
        str_schema('title', metadata: SealMetadata::filterable()->merge(SealMetadata::sortable())),
        float_schema('price'),
        datetime_schema('published_at'),
    ),
    index_name: 'products',
    identifier: 'id',
);

seal_schema_to_flow() does the reverse conversion, from a SEAL schema back to a Flow schema.

#Type Mapping

#Field Flags

You can override the default flags for each definition through its metadata:

Note: The index lifecycle (create and drop indexes) is a SEAL task. Use the engine directly: $engine->createIndex('users'), $engine->dropIndex('users').

Note: Some search backends index documents asynchronously. A document that the loader saved can be invisible to countDocuments() for a short time. Read the documentation of your backend. To wait for indexing tasks, SEAL has the ['return_slow_promise_result' => true] option.

Found a typo or an outdated section? Edit this page on GitHub ↗