Bytewax Streaming Pipeline Finished!

[iusztinpaul]

We are happy to announce that the streaming pipeline is finished. It is only working locally for now, but soon, we will show you how to deploy it.

Using it, you will use a web socket to listen to news from the Alpaca server constantly, use Bytewax to process the information, HuggingFace to embed the text, and Qdrant to load the embeddings to a vector DB.

To test it out, you have to:

make install
cp .env.example .env # ... and fill the env vars
make run

[akash28121992]

Hey can you tell the start of bytewax streaming how you have implemented it

[iusztinpaul]

Hello @akash28121992,

Sure.

We support x2 kinds of inputs. Here is how we build them inside the streaming_pipeline/flow.py file:

def _build_input(
    is_batch: bool = False,
    from_datetime: Optional[datetime.datetime] = None,
    to_datetime: Optional[datetime.datetime] = None,
    is_input_mocked: bool = False,
) -> Input:
    if is_input_mocked is True:
        return TestingInput(mocked.financial_news)

    if is_batch:
        assert (
            from_datetime is not None and to_datetime is not None
        ), "from_datetime and to_datetime must be provided when is_batch is True"

        return AlpacaNewsBatchInput(
            from_datetime=from_datetime, to_datetime=to_datetime, tickers=["*"]
        )
    else:
        return AlpacaNewsStreamInput(tickers=["*"])

AlpacaNewsBatchInput uses a RESTful API for historical data, and AlpacaNewsStreamInput uses a websocket for real-time data.

Feel free to let me know if you have any questions.

Also, we will soon release a full video lecture explaining how this works.