fix: time unit bug/dockerfile/add a guide (#44)

Signed-off-by: jyu6 <[email protected]>

Author: jy4096

Makefile

@@ -7,7 +7,7 @@ clean:
 	@find . -type f -name "*.py[co]" -exec rm -rf {} +
 
 format: clean
-	poetry run black pynumaflow/
+	poetry run black .
 
 
 lint: format

examples/developer_guide/Dockerfile

@@ -0,0 +1,54 @@
+####################################################################################################
+# builder: install needed dependencies
+####################################################################################################
+
+FROM python:3.10-slim-bullseye AS builder
+
+ENV PYTHONFAULTHANDLER=1 \
+  PYTHONUNBUFFERED=1 \
+  PYTHONHASHSEED=random \
+  PIP_NO_CACHE_DIR=on \
+  PIP_DISABLE_PIP_VERSION_CHECK=on \
+  PIP_DEFAULT_TIMEOUT=100 \
+  POETRY_VERSION=1.2.2 \
+  POETRY_HOME="/opt/poetry" \
+  POETRY_VIRTUALENVS_IN_PROJECT=true \
+  POETRY_NO_INTERACTION=1 \
+  PYSETUP_PATH="/opt/pysetup" \
+  VENV_PATH="/opt/pysetup/.venv"
+
+ENV PATH="$POETRY_HOME/bin:$VENV_PATH/bin:$PATH"
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends -y \
+        curl \
+        wget \
+        # deps for building python deps
+        build-essential \
+    && apt-get install -y git \
+    && apt-get clean && rm -rf /var/lib/apt/lists/* \
+    \
+    # install dumb-init
+    && wget -O /dumb-init https://github.com/Yelp/dumb-init/releases/download/v1.2.5/dumb-init_1.2.5_x86_64 \
+    && chmod +x /dumb-init \
+    && curl -sSL https://install.python-poetry.org | python3 -
+
+####################################################################################################
+# udf: used for running the udf vertices
+####################################################################################################
+FROM builder AS udf
+
+WORKDIR $PYSETUP_PATH
+COPY ./pyproject.toml ./poetry.lock ./
+RUN poetry install --no-cache --no-root && \
+    rm -rf ~/.cache/pypoetry/
+
+ADD . /app
+WORKDIR /app
+
+RUN chmod +x entry.sh
+
+ENTRYPOINT ["/dumb-init", "--"]
+CMD ["/app/entry.sh"]
+
+EXPOSE 5000

examples/developer_guide/Makefile

@@ -0,0 +1,3 @@
+.PHONY: image
+image:
+	docker build -t "test:latest" .

examples/developer_guide/README.md

@@ -0,0 +1,11 @@
+# Developer Guide
+
+This example is for numaflow-python contributors/developers. The example includes how to use your branch to build the UDF image to test your code change before submitting a PR. 
+
+1. Install poetry before starting your test. Make sure you have the correct python version.
+2. Push your code change to your branch.
+3. Update the `pynumaflow` in `pyproject.tomal` file with your (forked) repo url and your branch name. For example, `pynumaflow = {git = "https://github.com/chromevoid/numaflow-python", rev = "test-branch"}`
+4. Run `poetry update -vv` from this `developer_guide` folder. You should get a `poetry.lock` file.
+5. Update your `example.py` and the `image name` in `Makefile` as needed.
+6. Run `make image` to build your image.
+7. Now you have the image with your customized example and your code change to test in the numaflow pipeline. Example pipeline `pipeline-numaflow.yaml` is also provided in this `developer_guide` folder. Please check [numaflow](https://numaflow.numaproj.io/) for more details.

examples/developer_guide/entry.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+set -eux
+
+python example.py

examples/developer_guide/example.py

@@ -0,0 +1,38 @@
+from typing import Iterator
+from pynumaflow.function import (
+    Messages,
+    Message,
+    Datum,
+    Metadata,
+    UserDefinedFunctionServicer,
+)
+
+
+def map_handler(key: str, datum: Datum) -> Messages:
+    # forward a message
+    val = datum.value
+    _ = datum.event_time
+    _ = datum.watermark
+    messages = Messages()
+    messages.append(Message.to_vtx(key, val))
+    return messages
+
+
+def reduce_handler(key: str, datums: Iterator[Datum], md: Metadata) -> Messages:
+    # count the number of events
+    interval_window = md.interval_window
+    counter = 0
+    for _ in datums:
+        counter += 1
+    msg = (
+        f"counter:{counter} interval_window_start:{interval_window.start} "
+        f"interval_window_end:{interval_window.end}"
+    )
+    return Messages(Message.to_vtx(key, str.encode(msg)))
+
+
+if __name__ == "__main__":
+    grpc_server = UserDefinedFunctionServicer(
+        map_handler=map_handler, reduce_handler=reduce_handler
+    )
+    grpc_server.start()

examples/developer_guide/pipeline-numaflow.yaml

@@ -0,0 +1,42 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: pynumasdk
+spec:
+  vertices:
+    - name: in
+      source:
+        # A self data generating source
+        generator:
+          rpu: 10
+          duration: 1s
+    - name: cat
+      udf:
+        container:
+          # compute the sum
+          image: quay.io/numaio/numaflow-python/map-forward-message:latest
+    - name: counter
+      udf:
+        container:
+          # compute the sum
+          image: quay.io/numaio/numaflow-python/reduce-counter:latest
+        groupBy:
+          window:
+            fixed:
+              length: 10s
+          keyed: true
+    - name: sink
+      scale:
+        min: 1
+      sink:
+        udsink:
+          container:
+            image: quay.io/numaio/numaflow-python/sink-log:latest
+  edges:
+    - from: in
+      to: cat
+    - from: cat
+      to: counter
+      parallelism: 1
+    - from: counter
+      to: sink

examples/developer_guide/pyproject.toml

@@ -0,0 +1,15 @@
+[tool.poetry]
+name = "reduce-counter"
+version = "0.2.4"
+description = ""
+authors = ["Numaflow developers"]
+
+[tool.poetry.dependencies]
+python = "~3.10"
+pynumaflow = {git = "https://github.com/<YOUR_ID>/numaflow-python", rev = "<YOUR_BRANCH>"}
+
+[tool.poetry.dev-dependencies]
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"

examples/function/flatmap/Dockerfile

@@ -39,8 +39,8 @@ RUN apt-get update \
 FROM builder AS udf
 
 WORKDIR $PYSETUP_PATH
-COPY ./pyproject.toml ./poetry.lock ./
-RUN poetry install --without mlflowserver --no-cache --no-root && \
+COPY ./pyproject.toml ./
+RUN poetry install --no-cache --no-root && \
     rm -rf ~/.cache/pypoetry/
 
 ADD . /app

pynumaflow/function/server.py

@@ -152,8 +152,8 @@ def ReduceFn(
                 f"got key: {key}, start: {start}, end: {end}."
             )
 
-        start_dt = datetime.fromtimestamp(int(start), timezone.utc)
-        end_dt = datetime.fromtimestamp(int(end), timezone.utc)
+        start_dt = datetime.fromtimestamp(int(start) / 1e3, timezone.utc)
+        end_dt = datetime.fromtimestamp(int(end) / 1e3, timezone.utc)
         interval_window = IntervalWindow(start=start_dt, end=end_dt)
 
         try: