mirror of
https://github.com/Unstructured-IO/unstructured.git
synced 2025-07-08 01:22:43 +00:00
3eaf65a8c1
### Description This refactors the current ingest CLI process to support better granularity in how the steps are ran * Both multiprocessing and async now supported. Given that a lot of the steps are IO-bound, such as downloading and uploading content, we can achieve better parallelization by using async here * Destination step broken up into a stager step and an upload step. This will allow for steps that require manipulation of the data between formats, such as converting the elements json into a csv format to upload for tabular destinations, to be pulled out of the step that does the actual upload. * The process of writing the content to a local destination was now pulled out as it's own dedicated destination connector, meaning you no longer need to persist the content locally once the process is done if the content was uploaded elsewhere. * Quick update to the chunker/partition step to use the python client. * Move the uncompress suppport as a pipeline step since this can arbitrarily apply to any concrete files that have been downloaded, regardless of where they came from. * Leverage last modified date to mark files to be reprocessed, even if the file already exists locally. ### Callouts Retry configs haven't been moved over yet. This is an open question because the intent was for it to wrap potential connection errors but now any of the other steps that leverage an API might run into network connection issues. Should those be isolated in each of the steps and wrapped with the same retry configs? Or do we need to expose a unique retry config for each step? This would bloat the input params even more. ### Testing * If you want to run the new code as an SDK, there's an example file that was added to highlight how to do that: [example.py](https://github.com/Unstructured-IO/unstructured/blob/roman/refactor-ingest/unstructured/ingest/v2/example.py) * If you want to run the new code as an isolated CLI: ```shell PYTHONPATH=. python unstructured/ingest/v2/main.py --help ``` * If you want to see which commands have been migrated to the new version, there's now a `v2` short help text next to those commands when running the current cli: ```shell PYTHONPATH=. python unstructured/ingest/main.py --help Usage: main.py [OPTIONS] COMMAND [ARGS]...main.py --help Options: --help Show this message and exit. Commands: airtable azure biomed box confluence delta-table discord dropbox elasticsearch fsspec gcs github gitlab google-drive hubspot jira local v2 mongodb notion onedrive opensearch outlook reddit s3 v2 salesforce sftp sharepoint slack wikipedia ``` You can run any of the local or s3 specific ingest tests and these should now work. --------- Co-authored-by: ryannikolaidis <1208590+ryannikolaidis@users.noreply.github.com> Co-authored-by: rbiseck3 <rbiseck3@users.noreply.github.com>
36 lines
1.7 KiB
Bash
Executable File
36 lines
1.7 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
|
|
# Processes a the file from local, chunks, embeds, and writes the results to an Elasticsearch index.
|
|
|
|
# Structured outputs are stored in local-to-elasticsearch/
|
|
|
|
SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
|
|
cd "$SCRIPT_DIR"/../../.. || exit 1
|
|
|
|
# As an example we're using the local connector,
|
|
# however ingesting from any supported source connector is possible.
|
|
# shellcheck disable=2094
|
|
PYTHONPATH=. ./unstructured/ingest/main.py \
|
|
local \
|
|
--input-path example-docs/book-war-and-peace-1225p.txt \
|
|
--output-dir local-to-elasticsearch \
|
|
--strategy fast \
|
|
--chunking-strategy by_title \
|
|
--embedding-provider "<an unstructured embedding provider, ie. langchain-huggingface>" \
|
|
--num-processes 2 \
|
|
--verbose \
|
|
--work-dir "<directory for intermediate outputs to be saved>" \
|
|
elasticsearch \
|
|
--hosts "<List of URLs where elasticsearch index is served>" \
|
|
--index-name "<Index name to upload data in>" \
|
|
--username "<Username to authenticate into the index>" \
|
|
--password "<Password to authenticate into the index>" \
|
|
--batch-size-bytes "<Size limit for any batch to be uploaded, in bytes, ie. 15000000>" \
|
|
--num-processes "<Number of processes to be used to upload, ie. 2>" \
|
|
--cloud-id "<Id used to connect to Elastic Cloud>" \
|
|
--es-api-key "<Api key used for authentication>" \
|
|
--api-key-id "<Id associated with api key used for authentication: https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html>" \
|
|
--bearer-auth "<Bearer token used for HTTP bearer authentication>" \
|
|
--ca-certs "<path/to/ca/certs>" \
|
|
--ssl-assert-fingerprint "<SHA256 fingerprint value>"
|