Pathway

https://pathway.com/developers/
Admin
Pathway is a Python data processing framework for analytics and AI pipelines over data streams,...
Tokens:712,226
Snippets:6,189
Trust Score:9.4
Update:5 months ago
Show doc for...
Tokens:
Raw
### Test Persistence Setup (Python)

Source: https://pathway.com/developers/user-guide/deployment/persistence_recovery

Code to test the persistence setup by generating input data and starting the Pathway program in a separate process. This example simulates an interrupted run.

```python
# Clean the old files: remove old results and inputs
!rm -rf result.jsonlines
clean_input_directory()

# Start streaming inputs
pd_gen = multiprocessing.Process(
    target=generate_inputs,
     args=(0.05, 200, ["hello", "world"])
)
pd_gen.start()
```

--------------------------------

### Install Pathway

Source: https://pathway.com/developers/showcases/private-rag-ollama-mistral

Installs the Pathway library into a Python 3.10+ Linux runtime. Use this command for initial setup.

```bash
!pip install -U --prefer-binary pathway

```

--------------------------------

### Start Pathway Process on Machine 1

Source: https://pathway.com/developers/user-guide/advanced/running_on_multiple_machines

Use this command to start a Pathway process on machine 1, mirroring the setup of machine 0 with the same cluster addresses but a different process ID. The order of starting processes does not matter; the first to start will wait for others.

```bash
pathway spawn \
  --addresses 192.168.1.10:9000,192.168.1.11:9000 \
  --process-id 1 \
  python pipeline.py

```

--------------------------------

### Pathway Dockerfile Setup

Source: https://pathway.com/developers/templates/etl/realtime-log-monitoring

A Dockerfile to set up the Pathway environment. It uses a Python base image, installs Pathway and python-dateutil, and copies the alerts.py script.

```dockerfile
FROM --platform=linux/x86_64 python:3.10

RUN pip install -U pathway
RUN pip install python-dateutil
COPY ./pathway-src/alerts.py alerts.py

CMD ["python", "-u", "alerts.py"]
```

--------------------------------

### Install Pathway and Dependencies

Source: https://pathway.com/developers/templates/rag/_readmes/document_store_mcp_server

Install the Pathway library with all extras and project-specific requirements. This is necessary for running the template locally.

```bash
pip install pathway[all]
pip install -r requirements.txt

```

--------------------------------

### Install Pathway and OpenAI Packages

Source: https://pathway.com/developers/showcases/adaptive-rag

Installs the necessary Pathway and OpenAI Python packages. Uncomment the lines if you need to install them. Also downloads sample data for Adaptive RAG.

```bash
# Uncomment, if you need to install Pathway and OpenAI packages
# !pip install pathway
# !pip install openai
# Download `adaptive-rag-contexts.jsonl` with ~1000 contexts from SQUAD dataset
!wget -q -nc https://public-pathway-releases.s3.eu-central-1.amazonaws.com/data/adaptive-rag-contexts.jsonl

# If you want to use cache, set `PATHWAY_PERSISTENT_STORAGE environmental variable
# !export PATHWAY_PERSISTENT_STORAGE=".cache"
```

--------------------------------

### Complete JSON Lines read and write example

Source: https://pathway.com/developers/user-guide/connect/connectors/jsonlines-connector

A full example demonstrating reading from a JSON Lines file, writing to another JSON Lines file, and running the Pathway pipeline. This example copies streams of updates.

```python
import pathway as pw


class InputSchema(pw.Schema):
    key: int = pw.column_definition(primary_key=True)
    recipient: str
    sender: str


table = pw.io.jsonlines.read("./input_file.jsonl", schema=InputSchema)
pw.io.jsonlines.write(table, "./output_file.jsonl")
pw.run()
```

--------------------------------

### Install Pathway

Source: https://pathway.com/developers/templates/rag/private-rag-ollama-mistral

Installs the Pathway library into a Python 3.10+ Linux runtime. Use --prefer-binary for faster installation.

```bash
!pip install -U --prefer-binary pathway
```

--------------------------------

### Install Libraries

Source: https://pathway.com/developers/user-guide/temporal-data/asof-join

Installs the necessary Python libraries: pandas, yahooquery, and pathway. Use this command in your environment to set up the prerequisites.

```python
%%capture --no-display
# !pip install pandas yahooquery -U pathway

```

--------------------------------

### Install Pathway

Source: https://pathway.com/developers/templates/rag/multimodal-rag

Install Pathway and all its optional packages. Ensure you are using version 0.20.0 or higher.

```bash
!pip install 'pathway[all]>=0.20.0'
```

--------------------------------

### Install Pathway Locally

Source: https://pathway.com/developers/templates/rag/template-adaptive-rag

Install the Pathway library with all dependencies locally to run the application. This command should be executed in your terminal.

```bash
pip install pathway[all]

```

--------------------------------

### Create Input and Output Directories and Files

Source: https://pathway.com/developers/user-guide/connect/connectors/fs-connector

Prepares the basic file structure for the examples by creating directories and populating them with sample plaintext files.

```bash
! mkdir -p plain_input
! mkdir -p plain_output
! echo -e "test1\ndata1" > plain_input/in1.txt
! echo -e "test2\ndata2" > plain_input/in2.txt

```

--------------------------------

### Download Sample Data and Schema Files

Source: https://pathway.com/developers/user-guide/connect/schema-generation

Uncomment and execute this code to download necessary files for the tutorial. Ensure you have the necessary permissions and network access.

```python
%%capture --no-display
# !wget https://public-pathway-releases.s3.eu-central-1.amazonaws.com/data/schema-generation-sample-data.csv -O data.csv
# !wget https://public-pathway-releases.s3.eu-central-1.amazonaws.com/data/schema-generation-schema.json -O schema.json
```

--------------------------------

### Initializing DocumentStore

Source: https://pathway.com/developers/user-guide/llm-xpack/docs-indexing

Minimal example of how to initialize a DocumentStore with documents, a retriever factory, and a text splitter.

```APIDOC
## Initializing DocumentStore

### Description
This example shows the basic setup for a `DocumentStore`. It involves reading documents from a file system, defining a `TokenCountSplitter`, and then instantiating the `DocumentStore` with these components and a pre-configured `retriever_factory`.

### Code
```python
from pathway.xpacks.llm.document_store import DocumentStore
from pathway.xpacks.llm.splitters import TokenCountSplitter
import pathway as pw

# Assuming retriever_factory is already defined as shown in the previous example
# retriever_factory = BruteForceKnnFactory(...)

data_sources = pw.io.fs.read(
    "./sample_docs",
    format="binary",
    with_metadata=True,
)
text_splitter = TokenCountSplitter()
store = DocumentStore(
    docs=data_sources,
    retriever_factory=retriever_factory,
    splitter=text_splitter,
)
```
```

--------------------------------

### Install Poetry Dependencies

Source: https://pathway.com/developers/templates/rag/_readmes/drive_alert

Install project dependencies using Poetry, including optional extras for examples and unstructured data processing. This is a prerequisite for running services manually.

```bash
poetry install --with examples --extras unstructured
```

--------------------------------

### Run Pathway Monitoring Demo Script

Source: https://pathway.com/developers/templates/etl/_readmes/monitoring

Execute the example monitoring script. Ensure your license key and monitoring endpoint are correctly set within the script.

```bash
python monitoring_demo.py

```

--------------------------------

### Install Dependencies and Run App Manually

Source: https://pathway.com/developers/templates/rag/_readmes/unstructured_to_sql_on_the_fly

Install project dependencies using pip and then run the main application script. This starts the pipeline accessible via a REST API.

```bash
pip install -r requirements.txt
python app.py
```

--------------------------------

### Initialize Pub/Sub Publisher Client

Source: https://pathway.com/developers/api-docs/pathway-io-pubsub

Create a `pubsub_v1.PublisherClient` instance. This example shows initialization using service account credentials from a file.

```python
from google.cloud import pubsub_v1
publisher = pubsub_v1.PublisherClient.from_service_account_file(
    "./credentials.json"
)
```

--------------------------------

### Example Output of Exactly Once Behavior

Source: https://pathway.com/developers/user-guide/temporal-data/windows_with_behaviors

This table shows the output stream from the `exactly_once_behavior` example, illustrating window start and end times, the count of logs within each window, and associated timestamps.

```text
            | window_start | window_end | n_logs | __time__             | __diff__
^DX546QX... | 360          | 370        | 3      | 374                  | 1
^B5NPNV8... | 364          | 374        | 4      | 382                  | 1
^VRT1YTS... | 368          | 378        | 3      | 382                  | 1
^QE3WQ0X... | 372          | 382        | 2      | 18446744073709551614 | 1
^XJDDRJ0... | 376          | 386        | 3      | 18446744073709551614 | 1
^R0SM1CF... | 380          | 390        | 2      | 18446744073709551614 | 1
```

--------------------------------

### Launch Project Containers

Source: https://pathway.com/developers/templates/etl/_readmes/filebeat-pathway-slack

Use make commands to start all necessary Docker containers for the project. Connect to the Filebeat container to generate input streams.

```bash
make
make connect
./generate_input_stream.sh
```

--------------------------------

### Install Dependencies in Jupyter Notebook

Source: https://pathway.com/developers/user-guide/deployment/jupyter-docker

Example of installing Python packages directly within a Jupyter notebook code cell using the '!' prefix for shell commands. This approach is not suitable for Dockerfiles.

```python
!pip install langchain
!pip install langchain_community
!pip install lanchain_openai

```

--------------------------------

### Initialize Debug Table

Source: https://pathway.com/developers/api-docs/pathway-io/http

Sets up a sample table for demonstration purposes. This is a prerequisite for the following HTTP write examples.

```python
import pathway as pw
pets = pw.debug.table_from_markdown("owner pet \n Alice dog \n Bob cat \n Alice cat")
```

--------------------------------

### Example CSV Data

Source: https://pathway.com/developers/user-guide/deployment/persistence_restart_with_new_data

This is an example of CSV data representing user sessions, including session length, start and end times, event counts, and timestamps. It illustrates how session data might be structured.

```csv
user_id,session_length,session_start,session_finish,count,time,diff
"alice",583,1701103035,1701103618,5,1701270112954,-1
"alice",881,1701103035,1701103916,8,1701270112954,1
```

--------------------------------

### Performing Retrieval

Source: https://pathway.com/developers/user-guide/llm-xpack/docs-indexing

Example of how to use the retrieve_query function on a DocumentStore object to get relevant document chunks.

```APIDOC
## Performing Retrieval

### Description
After setting up the `DocumentStore` and preparing your queries, this code snippet shows the straightforward process of calling the `retrieve_query` method on the `store` object to obtain relevant document chunks based on your queries.

### Code
```python
# Assuming 'store' is an initialized DocumentStore object and 'query' is a DataFrame of queries
# result = store.retrieve_query(query)
# print(result) # To view the retrieval results
```
```

--------------------------------

### Download Data for Tutorial

Source: https://pathway.com/developers/templates/etl/suspicious_activity_tumbling_window

Uncomment and run this code to download the CSV file containing login data for the tutorial.

```python
# Uncomment to download the required files.
# %%capture --no-display
# !wget https://public-pathway-releases.s3.eu-central-1.amazonaws.com/data/suspicious_users_tutorial_logins.csv -O logins.csv

```

--------------------------------

### Run RAG App with Docker

Source: https://pathway.com/developers/templates/rag/_readmes/question_answering_rag

Start the RAG pipeline and UI by running the Docker compose setup.

```bash
docker compose up

```

--------------------------------

### Prepare Data Directory and File

Source: https://pathway.com/developers/user-guide/data-transformation/indexes-in-pathway

Creates a directory named 'points' and populates it with a CSV file containing sample point data (x, y coordinates).

```bash
mkdir points && echo -e "x,y\n2,2\n-2,2\n0,6\n2,8" > points/a.csv

```

--------------------------------

### Configure Filesystem Persistence Backend (Python)

Source: https://pathway.com/developers/user-guide/deployment/persistence_recovery

Sets up a persistence backend using the local filesystem. Specify the directory where snapshots will be stored.

```python
backend = pw.persistence.Backend.filesystem("./PStorage")
persistence_config = pw.persistence.Config(backend)
```

--------------------------------

### CSV Header Example

Source: https://pathway.com/developers/user-guide/connect/connectors/csv_connectors

Illustrates the required header format for CSV files used with Pathway connectors. Each file must start with a comma-separated list of column names.

```text
colA,colB

```

--------------------------------

### Spawn Pipeline with Initial Workers

Source: https://pathway.com/developers/user-guide/advanced/worker_count_scaling

Launch the pipeline using the `pathway spawn` command, specifying the initial number of workers with the `-n` flag. For example, to start with two workers:

```bash
pathway spawn -n 2 python pipeline.py
```

--------------------------------

### Configure HTTP Server

Source: https://pathway.com/developers/user-guide/llm-xpack/llm-app-pathway

Set up a lightweight HTTP server using `pw.io.http.PathwayWebserver` to receive user queries. Specify the host and port for the server.

```python
webserver = pw.io.http.PathwayWebserver(host="0.0.0.0", port=8011)
```

--------------------------------

### Create Test Table

Source: https://pathway.com/developers/api-docs/pathway-io/mqtt

Sets up a sample Pathway table for testing MQTT writes. Ensure Pathway is imported.

```python
import pathway as pw

table = pw.debug.table_from_markdown('''
age | owner | pet
10  | Alice | dog
9   | Bob   | cat
8   | Alice | cat
''')
```

--------------------------------

### Pathway Dockerfile

Source: https://pathway.com/developers/user-guide/connect/connectors/mongodb-debezium

A Dockerfile for setting up the Pathway container. It uses a Python 3.10 image, installs the Pathway library, copies the main script, and sets it as the command to run when the container starts.

```dockerfile
FROM --platform=linux/x86_64 python:3.10

RUN pip install -U pathway
COPY ./pathway-src/sum.py sum.py

CMD ["python", "-u", "sum.py"]

```

--------------------------------

### Initialize DocumentStore

Source: https://pathway.com/developers/user-guide/llm-xpack/docs-indexing

Create a DocumentStore to manage documents, splitting, and indexing. Requires data sources, a retriever factory, and a text splitter.

```python
from pathway.xpacks.llm.document_store import DocumentStore
from pathway.xpacks.llm.splitters import TokenCountSplitter
import pathway as pw

data_sources = pw.io.fs.read(
    "./sample_docs",
    format="binary",
    with_metadata=True,
)
text_splitter = TokenCountSplitter()
store = DocumentStore(
    docs=data_sources,
    retriever_factory=retriever_factory,
    splitter=text_splitter,
)
```

--------------------------------

### Create Milvus Collection Schema

Source: https://pathway.com/developers/api-docs/pathway-io/milvus

Before starting the pipeline, create the Milvus collection with a compatible schema. This example uses Milvus Lite and defines an integer primary key and a FLOAT_VECTOR field.

```python
import pathway as pw
from pymilvus import DataType, MilvusClient  

client = MilvusClient("./milvus.db")  
schema = client.create_schema(auto_id=False)  
schema.add_field("doc_id", DataType.INT64, is_primary=True)  
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=4)  
index_params = client.prepare_index_params()  
index_params.add_index("embedding", metric_type="COSINE", index_type="FLAT")  
client.create_collection("docs", schema=schema, index_params=index_params)  
client.close()
```

--------------------------------

### RAG Prompt Structure Example

Source: https://pathway.com/developers/showcases/adaptive-rag

This prompt template is used to guide the LLM in answering questions based on retrieved documents, including citation requirements and handling of unanswerable queries. It is inspired by the LlamaIndex QueryCitationEngine.

```text
Use the below articles to answer the subsequent question.
If the answer cannot be found in the articles, write
"I could not find an answer." Do not answer in full sentences.
When referencing information from a source, cite the appropriate
source(s) using their corresponding numbers. Every answer should
include at least one source citation.

Only cite a source when you are explicitly referencing it. For example:
"Source 1:
The sky is red in the evening and blue in the morning.
Source 2:
Water is wet when the sky is red.
Query: When is water wet?
Answer: When the sky is red [2], which occurs in the evening [1]."
Now it's your turn.


```

--------------------------------

### Create a Sample Pathway Table

Source: https://pathway.com/developers/api-docs/pathway-io/questdb

Demonstrates creating a simple Pathway table from markdown data, which will be used as the source for writing to QuestDB.

```python
import pathway as pw
table = pw.debug.table_from_markdown('''
     | data
   1 | Hello
   2 | World
''')

```

--------------------------------

### Add Intra-Machine Parallelism with Threads

Source: https://pathway.com/developers/user-guide/advanced/running_on_multiple_machines

Combine multi-machine setup with intra-machine parallelism by adding the `--threads` flag. This example configures two threads per machine for a total of four workers across two machines.

```bash
pathway spawn \
  --addresses 192.168.1.10:9000,192.168.1.11:9000 \
  --process-id 0 \
  --threads 2 \
  python pipeline.py

```

--------------------------------

### MCP Client Example

Source: https://pathway.com/developers/user-guide/llm-xpack/pathway_mcp_server

Use the `fastmcp` package to create a client that connects to the MCP server. It demonstrates listing available tools and calling a specific tool.

```python
import asyncio
from fastmcp import Client

PATHWAY_MCP_URL = "http://localhost:8123/mcp/"

client = Client(PATHWAY_MCP_URL)

async def main():
    async with client:
        tools = await client.list_tools()
        print(tools)

    async with client:
        result = await client.call_tool(name="get_constant_value", arguments={})
        print(result)

asyncio.run(main())
```

--------------------------------

### Subscribe to Table Changes with Callbacks

Source: https://pathway.com/developers/api-docs/pathway-io

Use this snippet to subscribe to a table and define callbacks for row changes and stream endings. Ensure `pw.run()` is called to start the computation. This example is incompatible with async callbacks due to the `sort_by` parameter not being used.

```python
import pathway as pw

table = pw.debug.table_from_markdown('\
     | pet  | owner   | age | __time__ | __diff__
   1 | dog  | Alice   | 10  | 0        | 1
   2 | cat  | Alice   | 8   | 2        | 1
   3 | dog  | Bob     | 7   | 4        | 1
   2 | cat  | Alice   | 8   | 6        | -1
')

def on_change(key: pw.Pointer, row: dict, time: int, is_addition: bool):
    print(f"{row}, {time}, {is_addition}")

def on_end():
    print("End of stream.")

pw.io.subscribe(table, on_change, on_end)
pw.run(monitoring_level=pw.MonitoringLevel.NONE)

```

--------------------------------

### Complete Pathway pipeline example

Source: https://pathway.com/developers/user-guide/introduction/concepts

This example demonstrates a complete Pathway pipeline, including Kafka connectors for input and output, data filtering, and aggregation. It shows how to define schemas, read from Kafka, process data, and write results back to Kafka, concluding with `pw.run()`.

```python
import pathway as pw


class InputSchema(pw.Schema):
    name: str = pw.column_definition(primary_key=True)
    age: int


input_table = pw.io.kafka.read(kafka_settings, schema=InputSchema, topic="topic1", format="json")
filtered_table = input_table.filter(input_table.age >= 0)
result_table = filtered_table.reduce(sum_age = pw.reducers.sum(filtered_table.age))
pw.io.kafka.write(result_table, kafka_settings, topic="topic2", format="json")

pw.run()

```

--------------------------------

### Launch Project Containers

Source: https://pathway.com/developers/templates/etl/_readmes/logstash-pathway-elastic

Use this command to start all necessary Docker containers for the project. Ensure you are in the root repository directory.

```bash
make
```

--------------------------------

### Docker Compose for Kafka and ZooKeeper

Source: https://pathway.com/developers/user-guide/connect/connectors/switching-to-redpanda

Defines services for ZooKeeper and Kafka in a Docker Compose setup. It configures Kafka to automatically create topics and connect to ZooKeeper. The command also ensures the 'ratings' topic is created before Kafka fully starts.

```yaml
version: "3.7"
name: tuto-switch-to-redpanda
networks:
  tutorial_network:
    driver: bridge
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.5.3
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    networks:
      - tutorial_network
  kafka:
    image: confluentinc/cp-enterprise-kafka:5.5.3
    depends_on: [zookeeper]
    environment:
      KAFKA_AUTO_CREATE_TOPICS: true
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9991
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      CONFLUENT_SUPPORT_METRICS_ENABLE: false
    ports:
    - 9092:9092
    command: sh -c "((sleep 15 && kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic ratings)&) && /etc/confluent/docker/run "
    networks:
      - tutorial_network

```

--------------------------------

### Start Multiple Processes on a Single Machine

Source: https://pathway.com/developers/user-guide/advanced/running_on_multiple_machines

When hosting multiple processes on a single machine, assign unique ports for each process. This example shows processes 0 and 1 running on '192.168.1.10' with different ports, while process 2 runs on another machine.

```bash
pathway spawn \
  --addresses 192.168.1.10:9000,192.168.1.10:9001,192.168.1.11:9000 \
  --process-id 0 \
  python pipeline.py

```

```bash
pathway spawn \
  --addresses 192.168.1.10:9000,192.168.1.10:9001,192.168.1.11:9000 \
  --process-id 1 \
  python pipeline.py

```

```bash
pathway spawn \
  --addresses 192.168.1.10:9000,192.168.1.10:9001,192.168.1.11:9000 \
  --process-id 2 \
  python pipeline.py

```

--------------------------------

### Connect to QuestDB with psql

Source: https://pathway.com/developers/api-docs/pathway-io/questdb

Example command to connect to the running QuestDB instance using `psql` to query the data.

```bash
psql -h localhost -p 8812 -U admin -d qdb

```

--------------------------------

### Expose Multiple Tools

Source: https://pathway.com/developers/user-guide/llm-xpack/pathway_mcp_server

Shows how to import necessary components and prepare for exposing multiple tools by defining each tool and adding them to the list to be served.

```python
import pathway as pw
from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp
```

--------------------------------

### Install Required Packages

Source: https://pathway.com/developers/templates/etl/option-greeks

Installs necessary Python packages for data manipulation and financial analysis. Ensure these are installed before running the code.

```python
!pip install pathway databento pandas scipy numpy python-dotenv

```

--------------------------------

### Set Up Streaming Data Source

Source: https://pathway.com/developers/templates/etl/live_data_jupyter

Create a streaming data source by replaying data from a CSV file using Pathway's demo replay functionality. This simulates a live data stream without requiring external infrastructure. The computational graph is built here and executed later.

```python
fname = "ticker.csv"
schema = pw.schema_from_csv(fname)
data = pw.demo.replay_csv(fname, schema=schema, input_rate=1000)

# # For static data exploration use
# data = pw.io.csv.read(fname, schema=schema, mode="static")

# Parse the timestamps
data = data.with_columns(t=data.t.dt.utc_from_timestamp(unit="ms"))
```

--------------------------------

### Install LLM xpack

Source: https://pathway.com/developers/user-guide/llm-xpack/overview

Install the LLM xpack functionality with Pathway using pip. You can install it as a separate package or as part of the 'all' option.

```bash
pip install "pathway[xpack-llm]"
```

```bash
pip install "pathway[all]"
```

--------------------------------

### Navigate to EL Pipeline Directory

Source: https://pathway.com/developers/templates/etl/el-pipeline

Change the directory to the EL pipeline example folder within the cloned Pathway repository.

```bash
cd examples/templates/el-pipeline/

```

--------------------------------

### Install Requirements

Source: https://pathway.com/developers/templates/rag/_readmes/multimodal_rag

Install necessary Python packages for the application.

```bash
pip install -r requirements.txt -U

```

--------------------------------

### Install Pathway with Extra Index URL

Source: https://pathway.com/developers/user-guide/introduction/installation

Installs the Pathway package by directly providing the extra index URL to the pip install command. This is an alternative to configuring pip.conf.

```bash
pip install -U pathway --extra-index-url=https://<username>:<password>@<URL>

```

--------------------------------

### Create Sample Log CSV Files

Source: https://pathway.com/developers/api-docs/pathway-io/csv

Creates a 'logs' directory and populates it with two CSV files, 'part_1.csv' and 'part_2.csv', containing site access log data.

```bash
mkdir logs
printf "id,ip,login\n1,127.0.0.1,alice\n2,8.8.8.8,alice" > logs/part_1.csv
printf "id,ip,login\n3,8.8.8.8,bob\n4,127.0.0.1,alice" > logs/part_2.csv
```

--------------------------------

### Dockerfile with Direct Pip Installs

Source: https://pathway.com/developers/user-guide/deployment/jupyter-docker

Install Python dependencies directly within the Dockerfile using 'RUN pip install' commands. This is suitable for a small number of dependencies.

```docker
FROM pathwaycom/pathway:latest

RUN pip install langchain
RUN pip install langchain_community
RUN pip install lanchain_openai

COPY . .

CMD [ "python", "./your-script.py" ]

```

--------------------------------

### Complete Document Store Configuration Example

Source: https://pathway.com/developers/user-guide/llm-xpack/pathway_mcp_server

A comprehensive YAML configuration for setting up a Pathway document store, including data sources, embedders, splitters, parsers, and retriever factories.

```yaml
$sources:
  - !pw.io.fs.read
    path: data
    format: binary
    with_metadata: true

$embedder: !pw.xpacks.llm.embedders.OpenAIEmbedder
  model: "text-embedding-ada-002"
  cache_strategy: !pw.udfs.DefaultCache {}

$splitter: !pw.xpacks.llm.splitters.TokenCountSplitter
  min_tokens: 250
  max_tokens: 600

$parser: !pw.xpacks.llm.parsers.DoclingParser {}

$knn_index: !pw.stdlib.indexing.BruteForceKnnFactory
  reserved_space: 1000
  embedder: $embedder
  metric: !pw.engine.BruteForceKnnMetricKind.COS

$bm25_index: !pw.stdlib.indexing.TantivyBM25Factory {}

$retriever_factory: !pw.stdlib.indexing.HybridIndexFactory
  retriever_factories:
    - $knn_index
    - $bm25_index

$document_store: !pw.xpacks.llm.document_store.DocumentStore
  docs: $sources
  parser: $parser
  splitter: $splitter
  retriever_factory: $retriever_factory

```

--------------------------------

### Install Pathway

Source: https://pathway.com/developers/user-guide/introduction/first_realtime_app_with_pathway

Install Pathway using pip for Python 3.10+.

```bash
pip install pathway

```

--------------------------------

### Set up HTTP REST Connector

Source: https://pathway.com/developers/user-guide/data-transformation/indexes-in-pathway

Initializes an HTTP REST connector to receive queries and a response writer. Configure host, port, schema, and autocommit settings.

```python
host = "0.0.0.0"  # set as needed
port = 8080  # set as needed
queries, response_writer = pw.io.http.rest_connector(
    host=host,
    port=port,
    schema=PointSchema,
    autocommit_duration_ms=50,
    delete_completed_queries=False,
)

```

--------------------------------

### Install Dependencies

Source: https://pathway.com/developers/templates/etl/_readmes/web-scraping

Installs all necessary Python packages listed in the requirements.txt file.

```bash
pip install -r requirements.txt

```
Tokens:
Raw