# Loading data > Start an indexing pipeline from files on disk, nodes you build in code, a tool executor, web pages, Kafka, Fluvio, Parquet or a custom Loader. Every indexing pipeline starts with a stream of nodes. You fill that stream from files, web pages, Kafka, Fluvio, Parquet or your own data, with a loader and `Pipeline::from_loader` or with any stream and `Pipeline::from_stream`. ## Files ```rust let languages = [SupportedLanguages::Rust, SupportedLanguages::Python]; let extensions: Vec<&str> = languages .iter() .flat_map(|language| language.file_extensions().iter().copied()) .collect(); // Extensions without the leading dot; each call adds to the list let loader = FileLoader::new("src") .with_extensions(&extensions) .with_extensions(&["md"]); Pipeline::from_loader(loader) .then_chunk(ChunkText::from_max_characters(2000)) .then_store_with(MemoryStorage::default()) .run() .await?; ``` Source: [`snippets/src/indexing/loading.rs#L17-L32`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/indexing/loading.rs#L17-L32) `FileLoader::new` takes a directory or a single file. Its walk uses the `ignore` crate's defaults. It skips hidden files and files excluded by `.ignore` files, and by `.gitignore` rules when the root is inside a git repository. `with_extensions` keeps only files with those extensions. Without any call the loader yields every file. `SupportedLanguages::file_extensions` returns the extensions tree-sitter knows for a language. It needs the `tree-sitter` feature flag. `FileLoader` itself needs none. Each file becomes one `TextNode`: - `chunk` holds the file content, with invalid UTF-8 replaced. - `path` is relative when the root is relative. The node id hashes path and chunk, so changing the root from relative to absolute misses the [node cache](/indexing/caching/). - A file that cannot be read becomes an error in the stream. The first error ends `run` unless you filter errors, see [Errors](/indexing/streams-and-concurrency/#errors). ## Nodes in memory ```rust let mut faq = TextNode::new("A loader turns a data source into a stream of nodes."); faq.path = "faq/loading.md".into(); faq.metadata.insert("Source", "faq"); let nodes = vec![ faq, TextNode::new("An indexing pipeline does nothing until you call run."), ]; let storage = MemoryStorage::default(); Pipeline::from_stream(nodes) .then_store_with(storage.clone()) .run() .await?; println!("Stored {} nodes", storage.get_all_values().await.len()); ``` Source: [`snippets/src/indexing/loading.rs#L40-L55`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/indexing/loading.rs#L40-L55) `Pipeline::from_stream` takes anything that converts into an `IndexingStream`. That includes a `Vec`, a boxed `Stream` of results and a Tokio `mpsc` receiver. A channel lets another task produce nodes while the pipeline runs: ```rust let (sender, receiver) = tokio::sync::mpsc::channel(100); tokio::spawn(async move { for line in ["first message", "second message"] { if sender.send(Ok(TextNode::new(line))).await.is_err() { break; } } // Dropping the sender ends the stream, and with it the run }); Pipeline::from_stream(receiver) .then_store_with(MemoryStorage::default()) .run() .await?; ``` Source: [`snippets/src/indexing/loading.rs#L63-L77`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/indexing/loading.rs#L63-L77) The stream ends when every sender is dropped. Until then, `run` keeps waiting for nodes. ## Files from a tool executor ```rust let executor = LocalExecutor::new("."); let files = executor .stream_files(Path::new("src"), Some(vec!["rs".to_string()])) .await?; Pipeline::from_stream(files) .then_chunk(ChunkText::from_max_characters(2000)) .then_store_with(MemoryStorage::default()) .run() .await?; ``` Source: [`snippets/src/indexing/loading.rs#L85-L95`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/indexing/loading.rs#L85-L95) `ToolExecutor::stream_files` returns the files below a path as an `IndexingStream`. `LocalExecutor` uses `path` as given and does not resolve it against the executor's working directory. `LocalExecutor` needs the `swiftide-agents` feature flag. ## Web pages ```rust title="examples/scraping_index_to_markdown.rs" indexing::Pipeline::from_loader(ScrapingLoader::from_spider( Website::new("https://www.bosun.ai/") .with_limit(1) .to_owned(), )) .then(HtmlToMarkdownTransformer::default()) .then_chunk(ChunkMarkdown::from_chunk_range(20..2048)) .log_all() .then_store_with(MemoryStorage::default()) .run() .await?; ``` Source: [`examples/scraping_index_to_markdown.rs#L26-L36`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/scraping_index_to_markdown.rs#L26-L36) `ScrapingLoader` crawls with the `spider` crate and needs the `scraping` feature flag. `ScrapingLoader::from_spider` takes a configured `Website`, here limited to one page. To configure it, add `spider` at the version swiftide uses (2.52), or the `Website` types do not match. Each node carries the raw HTML of one page, with the page URL as its `path`. `HtmlToMarkdownTransformer` converts the HTML to markdown and drops `script` and `style` tags. ### Spider Cloud Since 0.33, `ScrapingLoader` crawls through Spider Cloud when `SPIDER_CLOUD_API_KEY` is set and not empty. It reads these environment variables once per process: | Variable | Effect | | ---------------------- | ---------------------------------------------------------------------------------------------------------------------------------- | | `SPIDER_CLOUD_API_KEY` | Turns Spider Cloud on | | `SPIDER_CLOUD_MODE` | `proxy`, `api`, `unblocker` or `fallback`. Anything else selects smart mode, which uses the unblocker only against bot protection. | | `SPIDER_CLOUD_API_URL` | Optional API endpoint | Spider Cloud cannot reach hosts that only your machine can route to. A start URL on `localhost`, a `.localhost` or `.local` name, a loopback address, or a private or link-local IPv4 address is crawled directly. ## Kafka and Fluvio ```rust title="examples/kafka.rs" static LOADER_TOPIC: &str = "loader"; static STORAGE_TOPIC: &str = "storage"; let mut client_config = ClientConfig::new(); client_config.set("bootstrap.servers", "localhost:9092"); client_config.set("group.id", "group_id"); client_config.set("auto.offset.reset", "earliest"); let loader = Kafka::builder() .client_config(client_config.clone()) .topic(LOADER_TOPIC) .build() .unwrap(); let storage = Kafka::builder() .client_config(client_config) .topic(STORAGE_TOPIC) .create_topic_if_not_exists(true) .batch_size(2usize) .build() .unwrap(); ``` Source: [`examples/kafka.rs#L27-L47`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/kafka.rs#L27-L47) ```rust title="examples/kafka.rs" indexing::Pipeline::from_loader(loader) .then_in_batch(Embed::new(FastEmbed::try_default().unwrap()).with_batch_size(10)) .then_store_with(storage) .run() .await?; ``` Source: [`examples/kafka.rs#L51-L55`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/kafka.rs#L51-L55) The pipelines embed with `FastEmbed`, which needs the `fastembed` feature flag. `Kafka` needs the `kafka` feature flag, which builds librdkafka and needs cmake and a C toolchain. It takes an `rdkafka` `ClientConfig`, re-exported as `integrations::kafka::ClientConfig`. As a loader it subscribes to `topic` and turns each message payload into a node. Messages without a UTF-8 payload are skipped. The node metadata gets `kafka_topic`, `kafka_partition` and `kafka_offset`, plus `kafka_timestamp` and `kafka_key` when the message has them. `Kafka` is also a storage, here writing to a second topic. ```rust title="examples/fluvio.rs" static TOPIC_NAME: &str = "hello-rust"; static PARTITION_NUM: u32 = 0; let loader = Fluvio::builder() .consumer_config_ext( ConsumerConfigExt::builder() .topic(TOPIC_NAME) .partition(PARTITION_NUM) .offset_start(fluvio::Offset::from_end(1)) .build() .unwrap(), ) .build() .unwrap(); ``` Source: [`examples/fluvio.rs#L36-L49`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/fluvio.rs#L36-L49) `Fluvio` needs the `fluvio` feature flag and is a loader only. It takes a `ConsumerConfigExt` with the topic, partition and start offset. `Offset` comes from the `fluvio` crate, which you add at swiftide's version (0.50). Each record becomes a node with its key in the `fluvio_key` metadata field, which is null when the record has no key. A Kafka pipeline waits for new messages, so `run` does not return on its own. For both loaders, a receive error ends the run. ## Parquet ```rust let loader = Parquet::builder() .path("data.parquet") .column_name("chunk") .build()?; Pipeline::from_loader(loader) .then_chunk(ChunkText::from_max_characters(2000)) .then_store_with(MemoryStorage::default()) .run() .await?; ``` Source: [`snippets/src/indexing/loading.rs#L103-L112`](https://github.com/bosun-ai/swiftide-website/blob/master/snippets/src/indexing/loading.rs#L103-L112) `Parquet` needs the `parquet` feature flag. `batch_size` sets the rows read per batch and defaults to 1024. The loader reads one column and turns every non-null value into a node. The column must hold strings. A column of another type yields an error, and a missing column panics. Batches are flattened unordered, so node order can differ from row order. Each load prints the column descriptors to stderr. ## Custom loaders When no loader fits, implement `Loader`. The [docs-qa](/showcases/docs-qa/) showcase loads markdown and `llms-full.txt` sources as one node per page: ```rust title="examples/docs-qa/src/loader.rs" /// Yields one [`TextNode`] per documentation page, with the page title as metadata. #[derive(Debug, Clone)] pub struct DocsLoader { sources: Vec, } impl DocsLoader { pub fn new(sources: Vec) -> Self { Self { sources } } } impl Loader for DocsLoader { type Output = String; fn into_stream(self) -> IndexingStream { stream::iter(self.sources) .then(DocsSource::pages) .map_ok(|pages| stream::iter(pages).map(|page| Ok(page_node(page)))) .try_flatten() .boxed() .into() } // Only needed when the loader is used as a `Box`. fn into_stream_boxed(self: Box) -> IndexingStream { self.into_stream() } } fn page_node(page: Page) -> TextNode { let mut node = TextNode::new(page.markdown); node.path = page.origin.into(); node.metadata.insert(PAGE_TITLE, page.title.as_str()); node } ``` Source: [`examples/docs-qa/src/loader.rs#L74-L109`](https://github.com/bosun-ai/swiftide/blob/e377c4a2e76fe2559629ae0efdae3c6e0fb4d829/examples/docs-qa/src/loader.rs#L74-L109) - `type Output` is the chunk type of the nodes. Built-in transformers expect `String`. - `into_stream` is synchronous and consumes the loader. Put slow work inside the stream, like the page downloads this loader starts through `DocsSource::pages`, so it runs when the pipeline polls it. - A failing source becomes an `Err` item, not a panic. - `into_stream_boxed` is only needed when you use the loader as a `Box`. Its default implementation panics. - `Loader` requires `Send`, `Sync` and `DynClone`, which any `Clone` type implements. ## Gotchas - `Pipeline::from_loader` calls `into_stream` right away, so work starts when you build the pipeline, not when you call `run`. `Kafka` subscribes to its topic. `Fluvio` connects. `Parquet` opens the file. `ScrapingLoader` starts crawling, and panics outside a Tokio runtime. - `Fluvio` and `Parquet` use `tokio::task::block_in_place`, which panics on a current-thread runtime. `#[tokio::main]` uses a multi-thread runtime by default; `#[tokio::test]` does not, so use `#[tokio::test(flavor = "multi_thread")]`. - `Kafka`, `Fluvio` and `Parquet` panic when they cannot create their consumer or open their file. ## Feature flags Install the feature flags you need, for example: ```sh cargo add swiftide@0.34 --features tree-sitter,scraping,kafka,fluvio,parquet,fastembed ``` [Feature flags](/reference/feature-flags/) lists them all. ## Reference - [`FileLoader`](https://docs.rs/swiftide/0.34/swiftide/indexing/loaders/struct.FileLoader.html) - [`Loader`](https://docs.rs/swiftide/0.34/swiftide/traits/trait.Loader.html) - [`IndexingStream`](https://docs.rs/swiftide/0.34/swiftide/indexing/struct.IndexingStream.html) - [`Pipeline::from_loader`](https://docs.rs/swiftide/0.34/swiftide/indexing/struct.Pipeline.html#method.from_loader) - [`Pipeline::from_stream`](https://docs.rs/swiftide/0.34/swiftide/indexing/struct.Pipeline.html#method.from_stream) - [`TextNode`](https://docs.rs/swiftide/0.34/swiftide/indexing/type.TextNode.html) - [`SupportedLanguages`](https://docs.rs/swiftide/0.34/swiftide/integrations/treesitter/enum.SupportedLanguages.html) - [`ToolExecutor`](https://docs.rs/swiftide/0.34/swiftide/traits/trait.ToolExecutor.html) - [`LocalExecutor`](https://docs.rs/swiftide/0.34/swiftide/agents/tools/local_executor/struct.LocalExecutor.html) - [`ScrapingLoader`](https://docs.rs/swiftide/0.34/swiftide/integrations/scraping/struct.ScrapingLoader.html) - [`HtmlToMarkdownTransformer`](https://docs.rs/swiftide/0.34/swiftide/integrations/scraping/struct.HtmlToMarkdownTransformer.html) - [`Kafka`](https://docs.rs/swiftide/0.34/swiftide/integrations/kafka/struct.Kafka.html) - [`Fluvio`](https://docs.rs/swiftide/0.34/swiftide/integrations/fluvio/struct.Fluvio.html) - [`Parquet`](https://docs.rs/swiftide/0.34/swiftide/integrations/parquet/struct.Parquet.html) ## Next [Chunking](/indexing/chunking/)