### Source Vertex Example Calculation Source: https://github.com/numaproj/numaflow/blob/main/docs/specifications/autoscaling.md Example demonstrating the calculation of desired replicas for a source vertex. ```plaintext desiredReplicas = 60000 / (3 * (10000 / 2)) = 4 ``` -------------------------------- ### Build and Install Numaflow Controller Source: https://github.com/numaproj/numaflow/blob/main/docs/development/development.md Builds the Numaflow source code, creates a container image, and installs the controller into the 'numaflow-system' namespace. ```shell make start ``` -------------------------------- ### Group Examples Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/ui/authz/rbac.md Provides examples of adding users to predefined RBAC groups, enabling consolidated permission management. ```plaintext g, test@test.com, role:readonly g, test_user, role:admin ``` -------------------------------- ### Install kind and Create Kubernetes Cluster Source: https://github.com/numaproj/numaflow/blob/main/docs/development/development.md Installs the 'kind' tool on macOS and creates a local Kubernetes cluster. It also exports the kubeconfig for the newly created cluster. ```shell # Install kind on macOS brew install kind # Create a cluster with default name kind kind create cluster # Get kubeconfig for the cluster kind export kubeconfig ``` -------------------------------- ### Example Sliding Window Configuration Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/windowing/sliding.md An example of a sliding window configuration for a reduce vertex, setting a window length of 60 seconds and a slide interval of 10 seconds. ```yaml vertices: - name: my-udf udf: groupBy: window: sliding: length: 60s slide: 10s ``` -------------------------------- ### Add Init Container to UDF Vertex Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/init-containers.md Configure an init container for a UDF vertex to perform setup tasks before the main function container starts. This example uses a busybox image to echo a message and sleep. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-udf initContainers: - name: my-init image: busybox:latest command: ["/bin/sh", "-c", "echo \"my-init is running!\" && sleep 60"] udf: container: image: my-function:latest ``` -------------------------------- ### Verify ISB Service Installation Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/map/examples.md Verifies that the ISB Service is running and its pods are ready. This confirms the successful installation of the data buffering service. ```shell kubectl get isbsvc # Wait for pods to be ready $ kubectl get pods NAME READY STATUS RESTARTS AGE isbsvc-default-js-0 3/3 Running 0 19s isbsvc-default-js-1 3/3 Running 0 19s isbsvc-default-js-2 3/3 Running 0 19s ``` -------------------------------- ### Sink Example with Detailed Retry Strategy Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sinks/retry-strategy.md Illustrates a complete sink configuration including a detailed retry strategy with specific backoff parameters and a fallback sink. This example shows how to configure retries for a primary sink and route failed messages to a fallback sink if retries are exhausted. ```yaml sink: retryStrategy: backoff: interval: 500ms steps: 10 factor: 2.2 cap: 10s onFailure: 'fallback' udsink: container: image: my-sink-image fallback: udsink: container: image: my-fallback-sink ``` -------------------------------- ### Configure Init Container Resources Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/container-resources.md Specify resource limits and requests for the 'init' init-container of vertex pods. This is used for setup tasks before the main container starts. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-vertex initContainerTemplate: resources: limits: cpu: "2" memory: 2Gi requests: cpu: "1" memory: 1Gi ``` -------------------------------- ### Keyed vs. Non-Keyed Window Configuration Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/windowing/windowing.md Example demonstrating how to configure a vertex for keyed windowing. The `partitions` field is optional and defaults to 1, while `keyed` defaults to false. ```yaml vertices: - name: my-reduce partitions: 5 # Optional, defaults to 1 udf: groupBy: window: ... keyed: true # Optional, defaults to false ``` -------------------------------- ### Verify ISB Service Installation Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/examples.md Verifies the installation of the ISB Service by checking its status and the readiness of its associated pods. This confirms that the service is running and ready to handle data. ```shell kubectl get isbsvc NAME TYPE PHASE MESSAGE AGE default jetstream Running 3d19h # Wait for pods to be ready $ kubectl get pods NAME READY STATUS RESTARTS AGE isbsvc-default-js-0 3/3 Running 0 19s isbsvc-default-js-1 3/3 Running 0 19s isbsvc-default-js-2 3/3 Running 0 19s ``` -------------------------------- ### Serve Generated Numaflow Documentation Locally Source: https://github.com/numaproj/numaflow/blob/main/docs/development/development.md Starts a local HTTP server on port 8000 to host the generated Numaflow documentation, allowing for local preview. ```shell make docs-serve ``` -------------------------------- ### Install Numaflow Components Source: https://github.com/numaproj/numaflow/blob/main/docs/quick-start.md Applies the Numaflow installation manifest to the Kubernetes cluster within the specified namespace. Ensure the namespace exists before running this command. ```shell kubectl apply -n numaflow-system -f https://raw.githubusercontent.com/numaproj/numaflow/main/config/install.yaml ``` -------------------------------- ### Mounting ConfigMaps to Vertices Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/volumes.md Example demonstrating how to mount a ConfigMap to a udsource, udf, and udsink vertex. Ensure the volume name in the vertex definition matches the name in the volumeMounts. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-source volumes: - name: my-udsource-config configMap: name: udsource-config source: udsource: container: image: my-source:latest volumeMounts: - mountPath: /path/to/my-source-config name: my-udsource-config - name: my-udf volumes: - name: my-udf-config configMap: name: udf-config udf: container: image: my-function:latest volumeMounts: - mountPath: /path/to/my-function-config name: my-udf-config - name: my-sink volumes: - name: my-udsink-config configMap: name: udsink-config sink: udsink: container: image: my-sink:latest volumeMounts: - mountPath: /path/to/my-sink-config name: my-udsink-config ``` -------------------------------- ### Conditional Forwarding Examples Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/conditional-forwarding.md Illustrates different conditional forwarding scenarios using 'or', 'not', and 'and' operators for tag-based routing. ```yaml edges: - from: p1 to: even-vertex conditions: tags: operator: or # Optional, defaults to "or". values: - even-tag - from: p1 to: odd-vertex conditions: tags: operator: not values: - odd-tag - from: p1 to: all conditions: tags: operator: and values: - odd-tag - even-tag ``` -------------------------------- ### Install ISB Service Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/map/examples.md Installs the Inter-Step Buffer Service using a Kubernetes manifest. This service is essential for passing data between pipeline vertices. ```shell kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/stable/examples/0-isbsvc-jetstream.yaml ``` -------------------------------- ### Kafka SSL Configuration Properties Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/kafka.md Example configuration file for connecting to Kafka using SSL. Ensure you provide the correct path to your JKS truststore and its password. ```properties request.timeout.ms=20000 bootstrap.servers= security.protocol=SSL ssl.enabled.protocols=TLSv1.2 ssl.truststore.location= ssl.truststore.password= ``` -------------------------------- ### Configure Namespace Scope Installation via ConfigMap Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/installation.md Sets up Numaflow for namespace-scoped installation by configuring the 'numaflow-cmd-params-config' ConfigMap. This ensures Numaflow only watches pipelines within its own namespace. ```yaml apiVersion: v1 kind: ConfigMap metadata: name: numaflow-cmd-params-config data: # Whether to run in namespaced scope, defaults to false. namespaced: "true" ``` -------------------------------- ### Pulsar Source Concurrency Constraint Example Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/mvtx-streaming.md This example demonstrates how to configure the `maxUnack` setting on a Pulsar source to respect the `spec.limits.concurrency` of the MVTX. Ensure `spec.limits.concurrency` is less than or equal to the Pulsar source's `maxUnack`. ```yaml spec: streaming: true limits: concurrency: 800 # must be <= pulsar source maxUnack (default 1000) source: pulsar: maxUnack: 1000 # ... ``` -------------------------------- ### Numaflow Namespace Scope Installation with Kustomize Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/installation.md Configures Kustomize for a namespace-scoped Numaflow installation using minimal CRDs and the namespaced controller. Specify the desired version using the 'ref' parameter. ```yaml apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - https://github.com/numaproj/numaflow/config/advanced-install/minimal-crds?ref=stable # Or specify a version - https://github.com/numaproj/numaflow/config/advanced-install/namespaced-controller?ref=stable # Or specify a version namespace: numaflow-system ``` -------------------------------- ### Policy Examples Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/ui/authz/rbac.md Illustrates various ways to define RBAC policies, including wildcard usage for broad access and specific user/group permissions. ```plaintext p, test@test.com, *, *, POST p, test_user, *, *, * p, role:admin_ns, test_ns, *, * p, test_user, test_ns, *, GET ``` -------------------------------- ### Default Container Resources Configuration Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/controller-configmap.md Example of configuring default container resources (limits and requests for memory and CPU) for steps across all pipelines within the controller ConfigMap. ```yaml apiVersion: v1 kind: ConfigMap metadata: name: numaflow-controller-config data: controller-config.yaml: | defaults: containerResources: | limits: memory: "256Mi" cpu: "200m" requests: memory: "128Mi" cpu: "100m" ``` -------------------------------- ### Enable Tracing for All Vertices Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/tracing.md Apply tracing configuration to all vertex pods in a pipeline by setting environment variables in the `containerTemplate`. This example configures the OTLP traces endpoint and sampling. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: templates: vertex: containerTemplate: env: - name: OTEL_EXPORTER_OTLP_TRACES_ENDPOINT value: "http://otel-collector.observability.svc.cluster.local:4317" - name: OTEL_TRACES_SAMPLER value: "parentbased_traceidratio" - name: OTEL_TRACES_SAMPLER_ARG value: "0.001" vertices: - name: in source: generator: rpu: 1 duration: 1s - name: cat udf: container: image: quay.io/numaio/numaflow-go/map-cat:stable - name: out sink: log: {} edges: - from: in to: cat - from: cat to: out ``` -------------------------------- ### Configure MonoVertex Autoscaling Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/autoscaling.md Example of configuring autoscaling parameters for a MonoVertex. Adjust settings like min/max replicas, cooldown periods, and target processing seconds to optimize performance. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: MonoVertex metadata: name: my-mvtx spec: scale: disabled: false # Optional, defaults to false. min: 0 # Optional, minimum replicas, defaults to 0. max: 20 # Optional, maximum replicas, defaults to 50. lookbackSeconds: 120 # Optional, defaults to 120. scaleUpCooldownSeconds: 90 # Optional, defaults to 90. scaleDownCooldownSeconds: 90 # Optional, defaults to 90. zeroReplicaSleepSeconds: 120 # Optional, defaults to 120. targetProcessingSeconds: 20 # Optional, defaults to 20. replicasPerScaleUp: 2 # Optional, defaults to 2. replicasPerScaleDown: 2 # Optional, defaults to 2. ``` -------------------------------- ### Kafka SASL/PLAIN Configuration Properties Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/kafka.md Example configuration file for connecting to Kafka using SASL/PLAIN authentication over SSL. Replace placeholders with your actual credentials and server details. ```properties ssl.endpoint.identification.algorithm=https sasl.mechanism=PLAIN request.timeout.ms=20000 bootstrap.servers= retry.backoff.ms=500 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="" \ password="" security.protocol=SASL_SSL ``` -------------------------------- ### MonoVertex with RollingUpdate Strategy Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/update-strategy.md Example of a MonoVertex resource configured with the RollingUpdate strategy. This shows how to set `maxUnavailable` to an absolute number for a MonoVertex. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: MonoVertex metadata: name: my-mvtx spec: source: udsource: container: image: my-image1 sink: udsink: container: image: my-image2 updateStrategy: rollingUpdate: maxUnavailable: 2 type: RollingUpdate ``` -------------------------------- ### Install Validating Webhook Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/validating-webhook.md Apply the validating webhook configuration to the numaflow-system namespace using kubectl. ```shell kubectl apply -n numaflow-system -f https://raw.githubusercontent.com/numaproj/numaflow/stable/config/validating-webhook-install.yaml ``` -------------------------------- ### Pipeline with RollingUpdate Strategy Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/update-strategy.md Example of a Pipeline resource configured with the RollingUpdate strategy. This demonstrates how to specify `maxUnavailable` for a vertex within a Pipeline. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: simple-pipeline spec: vertices: - name: my-vertex updateStrategy: rollingUpdate: maxUnavailable: 25% type: RollingUpdate ``` -------------------------------- ### User-Defined Sink Vertex Configuration Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sinks/user-defined-sinks.md Example YAML configuration for a user-defined sink vertex, specifying the container image to use. ```yaml spec: vertices: - name: output sink: udsink: container: image: my-sink:latest ``` -------------------------------- ### GO_BACK_N Mode: Slope Calculation Example Source: https://github.com/numaproj/numaflow/blob/main/rust/numaflow-throttling/DESIGN.md Illustrates the calculation of the 'slope' parameter for the GO_BACK_N mode. This slope determines the rate at which the token pool size is adjusted based on usage and duration. ```plaintext slope = (max_tokens - min_tokens) / duration ``` -------------------------------- ### Configure Numaflow Pipeline Autoscaling Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/autoscaling.md This example demonstrates how to configure autoscaling parameters for a Numaflow Pipeline. You can adjust settings like minimum/maximum replicas, cooldown periods, and scaling targets. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-vertex scale: disabled: false # Optional, defaults to false. min: 0 # Optional, minimum replicas, defaults to 0. max: 20 # Optional, maximum replicas, defaults to 50. lookbackSeconds: 120 # Optional, defaults to 120. scaleUpCooldownSeconds: 90 # Optional, defaults to 90. scaleDownCooldownSeconds: 90 # Optional, defaults to 90. zeroReplicaSleepSeconds: 120 # Optional, defaults to 120. targetProcessingSeconds: 20 # Optional, defaults to 20. targetBufferAvailability: 50 # Optional, defaults to 50. replicasPerScaleUp: 2 # Optional, defaults to 2. replicasPerScaleDown: 2 # Optional, defaults to 2. --- ``` -------------------------------- ### Example MVTX Spec with Streaming Enabled Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/mvtx-streaming.md A sample MVTX specification enabling streaming mode with configured read batch size and concurrency limits, and specifying a JetStream source and a UDSink. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: MonoVertex metadata: name: streaming-mvtx spec: streaming: true limits: readBatchSize: 50 concurrency: 200 source: jetstream: # ... sink: udsink: container: image: my-sink:latest ``` -------------------------------- ### Configure Sampling Rate to 1% Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/tracing.md Set the sampling rate to 1% by configuring the OTEL_TRACES_SAMPLER to 'parentbased_traceidratio' and OTEL_TRACES_SAMPLER_ARG to '0.01'. This example shows how to set these environment variables within a container template. ```yaml containerTemplate: env: - name: OTEL_EXPORTER_OTLP_TRACES_ENDPOINT value: "http://otel-collector.observability.svc.cluster.local:4317" - name: OTEL_TRACES_SAMPLER value: "parentbased_traceidratio" - name: OTEL_TRACES_SAMPLER_ARG value: "0.01" # 1% ``` -------------------------------- ### Java Transformer: Event Time Filter Setup Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/transformer/overview.md This Java code snippet shows the beginning of a Numaflow SourceTransformer class for filtering events by time. It defines constants for specific dates and initializes a server. The full implementation of the server start and the transformer logic would follow. ```java public class EventTimeFilterFunction extends SourceTransformer { private static final Instant januaryFirst2022 = Instant.ofEpochMilli(1640995200000L); private static final Instant januaryFirst2023 = Instant.ofEpochMilli(1672531200000L); public static void main(String[] args) throws Exception { Server server = new Server(new EventTimeFilterFunction()); // Start the server server.start(); ``` -------------------------------- ### Attaching Persistent Volume Claim (PVC) Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/volumes.md Example showing how to attach a Persistent Volume Claim (PVC) to a udsource container. The volume name must be consistent between the vertex definition and the volumeMounts. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-source volumes: - name: mypd persistentVolumeClaim: claimName: myclaim source: udsource: container: image: my-source:latest volumeMounts: - mountPath: /path/to/my-source-config name: mypd ``` -------------------------------- ### Port-forwarding and Sending Data to HTTP Source Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/http.md Demonstrates how to test an HTTP source by port-forwarding from a local machine to the Vertex Pod and then sending a POST request with data. ```sh kubectl port-forward pod ${pod-name} 8443 curl -kq -X POST -d "hello world" https://localhost:8443/vertices/in ``` -------------------------------- ### Install Numaflow in Cluster Scope Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/installation.md Installs the latest stable Numaflow in cluster scope using kubectl. Ensure you are in the correct namespace. ```shell kubectl apply -n numaflow-system -f https://raw.githubusercontent.com/numaproj/numaflow/stable/config/install.yaml ``` -------------------------------- ### Side Input Directory and File Paths Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/side-inputs.md Illustrates the constant directory path for side inputs and the pattern for individual side input file names. ```Go sideinput.DirPath -> "/var/numaflow/side-inputs" sideInputFileName -> "/var/numaflow/side-inputs/sideInputName" ``` -------------------------------- ### Example Numaflow Pipeline Spec Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/kustomize/kustomize.md This is an example of a Numaflow Pipeline specification, including source, UDF, and sink vertices, along with defined edges. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: in source: generator: rpu: 5 duration: 1s - name: my-udf udf: container: image: my-pipeline/my-udf:v0.1 - name: out sink: log: {} edges: - from: in to: my-udf - from: my-udf to: out ``` -------------------------------- ### Numaflow Cluster Scope Installation with Kustomize Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/installation.md Configures Kustomize for a cluster-scoped Numaflow installation. Specify the desired version using the 'ref' parameter. ```yaml apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - https://github.com/numaproj/numaflow/config/cluster-install?ref=stable # Or specify a version namespace: numaflow-system ``` -------------------------------- ### GO_BACK_N Mode: Sample Data (No Gaps) Source: https://github.com/numaproj/numaflow/blob/main/rust/numaflow-throttling/DESIGN.md Sample data demonstrating token pool behavior in GO_BACK_N mode without gaps in calls. Observe how token pool size fluctuates based on token usage relative to the threshold. ```plaintext token_pool | tokens_used | ------------|-------------| 10 | 10 | 20 | 10 | 10 | 10 | 20 | 10 | 10 | 10 | 20 | 20 | 30 | 20 | 20 | 20 | 30 | 20 | 20 | 20 | 30 | 30 | 40 | 30 | 30 | 30 | 40 | 30 | 30 | 20 | 20 | 20 | 30 | 30 | 40 | 40 | 50 | 50 | 60 | 50 | ``` -------------------------------- ### Build and Deploy Numaflow UI Source: https://github.com/numaproj/numaflow/blob/main/ui/README.md Build the UI image and deploy it to a k3d cluster. Then, port-forward the service to access the UI locally. ```shell # Build image and deploy to a k3d cluster make start # Port-forward, and access https://localhost:8443 kubectl -n numaflow-system port-forward svc/numaflow-server 8443 ``` -------------------------------- ### Example Output of Fixed Window Sum Pipeline Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/examples.md Shows the expected output from the fixed window sum pipeline. It displays the aggregated payload, key, and window start/end times for each processed window. ```text 2023/01/05 11:54:41 (sink) Payload - 300 Key - odd Start - 60000 End - 120000 2023/01/05 11:54:41 (sink) Payload - 600 Key - even Start - 60000 End - 120000 2023/01/05 11:54:41 (sink) Payload - 300 Key - odd Start - 120000 End - 180000 2023/01/05 11:54:41 (sink) Payload - 600 Key - even Start - 120000 End - 180000 2023/01/05 11:54:42 (sink) Payload - 600 Key - even Start - 180000 End - 240000 2023/01/05 11:54:42 (sink) Payload - 300 Key - odd Start - 180000 End - 240000 ``` -------------------------------- ### GO_BACK_N Mode Configuration Source: https://github.com/numaproj/numaflow/blob/main/rust/numaflow-throttling/DESIGN.md Configure the GO_BACK_N mode with threshold percentage, cooldown period, and ramp-down percentage. Ensure the threshold percentage is met to increase the token pool; otherwise, it decreases. ```yaml modes: goBackN: thresholdPercentage: 50 # at least 50% of tokens should be used before token pool is increased, otherwise decreased coolDownPeriod: "5s" # After more than 5s of no calls to the rate limiter, the token pool size is reduced rampDownPercentage: 50 # The % of slope by which the token pool size is reduced ``` -------------------------------- ### Create and Push Release Tag Source: https://github.com/numaproj/numaflow/blob/main/docs/development/releasing.md After preparing the release, run this command to finalize the release and push the new tag. GitHub actions will automatically build and publish the new release. Replace `{x.y.z}` with the target version. ```shell make release VERSION=v{x.y.z} ``` -------------------------------- ### Deploy Simple MonoVertex Source: https://github.com/numaproj/numaflow/blob/main/docs/quick-start.md Deploy a simple MonoVertex containing a User-Defined Source, Transformer, and User-Defined Sink (log sink). ```shell kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/main/examples/21-simple-mono-vertex.yaml ``` -------------------------------- ### Install Minimal CRDs and Namespaced Controller Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/installation.md Applies minimal CRD definitions and a namespaced controller for Numaflow. This approach is recommended for namespace-scoped installations to avoid backward compatibility issues during upgrades. ```shell # Minimal CRD kubectl apply -f https://raw.githubusercontent.com/numaproj/numaflow/main/config/advanced-install/minimal-crds.yaml # Controller in namespaced scope kubectl apply -n numaflow-system -f https://raw.githubusercontent.com/numaproj/numaflow/stable/config/advanced-install/namespaced-controller-wo-crds.yaml ``` -------------------------------- ### Sprig Contains Function Example Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/map/builtin-functions/filter.md This example demonstrates using the Sprig 'contains' function to check if a specific string is present within a JSON field of the payload. It checks if 'James' is contained in the 'name' field. ```expression sprig.contains('James', json(payload).name) ``` -------------------------------- ### Resume a MonoVertex with Minimal Pods Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/mvtx-operations.md Execute this command to resume a MonoVertex using the minimum number of pods defined in 'spec.scale.min', or 1 if not specified. ```bash kubectl patch mvtx my-mvtx --type=merge --patch '{"spec": {"lifecycle": {"desiredPhase": "Running"}, "replicas": null}}' ``` -------------------------------- ### Configure Jetstream Source Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/jetstream.md Example configuration for a Jetstream source, specifying connection URL, stream, and optional consumer settings like deliver policy and subject filtering. TLS and authentication options are also shown. ```yaml spec: vertices: - name: input source: jetstream: url: nats://demo.nats.io # Jetstream server url stream: my-stream # stream name consumer: my-consumer # Optional, Name of the consumer on the stream. deliver_policy: last_per_subject # Optional, The point in the stream from which to receive messages. Defaults to "all" filter_subjects: # Optional, A set of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers - "abc.A.*" - "abc.B.*" tls: # Optional. insecureSkipVerify: # Optional, whether to skip TLS verification. Default to false. caCertSecret: # Optional, a secret reference, which contains the CA Cert. name: my-ca-cert key: my-ca-cert-key certSecret: # Optional, pointing to a secret reference which contains the Cert. name: my-cert key: my-cert-key keySecret: # Optional, pointing to a secret reference which contains the Private Key. name: my-pk key: my-pk-key auth: # Optional. basic: # Optional, pointing to the secret references which contain user name and password. user: name: my-secret key: my-user password: name: my-secret key: my-password ``` -------------------------------- ### Filter Expression Example Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/map/builtin-functions/filter.md This example demonstrates how to use the filter built-in function with an expression to filter messages based on the 'id' field. It converts the payload to JSON, extracts the 'id', and checks if it's less than 100. ```yaml - name: filter-vertex udf: builtin: name: filter kwargs: expression: int(json(payload).id) < 100 ``` -------------------------------- ### Sprig Base64 Decode Example Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/map/builtin-functions/filter.md This example shows how to use Sprig's base64 decode function within a filter expression. It decodes the payload, converts it to JSON, extracts the 'id', and checks if it's less than 100. ```expression int(json(sprig.b64dec(payload)).id) < 100 ``` -------------------------------- ### StartArgs Source: https://github.com/numaproj/numaflow/blob/main/docs/APIs.md Optional arguments to start the nats-server, such as enabling debugging. ```APIDOC ## StartArgs ### Description Optional arguments to start the nats-server, such as enabling debugging. ### Fields - **startArgs** (string array) - Optional - Arguments to start nats-server (e.g., "-D", "-DV"). ``` -------------------------------- ### Reduce Storage with emptyDir (for experimentation) Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/reduce.md Use emptyDir for temporary storage with reduce UDFs, suitable for experimentation. Note that this configuration leads to data loss upon pod migration and is not recommended for production. ```yaml vertices: - name: my-udf udf: groupBy: storage: emptyDir: {} ``` -------------------------------- ### Disable Autoscaling for MonoVertex Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/autoscaling.md Example of disabling autoscaling for a MonoVertex by setting `disabled: true`. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: MonoVertex metadata: name: my-mvtx spec: scale: disabled: true ``` -------------------------------- ### Define 60-Second Fixed Window Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/windowing/fixed.md This example demonstrates defining a fixed window with a length of 60 seconds. The window boundaries are determined by rounding down from the current time to the nearest multiple of the window length. ```yaml vertices: - name: my-udf udf: groupBy: window: fixed: length: 60s ``` -------------------------------- ### Get Message Path Info Source: https://github.com/numaproj/numaflow/blob/main/docs/core-concepts/serving.md Retrieve information about the message path for a given request ID. ```APIDOC ## GET /v1/process/message ### Description Retrieves information related to the message path for a specific request ID. ### Method GET ### Endpoint /v1/process/message ### Query Parameters - **id** (string) - Required - The unique identifier of the request for which to get message path info. ### Response #### Success Response (200) - **messagePathInfo** (object) - Details about the message path. ``` -------------------------------- ### Create Numaflow Namespace Source: https://github.com/numaproj/numaflow/blob/main/docs/quick-start.md Creates a dedicated namespace for Numaflow system components. This is a prerequisite for installing Numaflow. ```shell kubectl create ns numaflow-system ``` -------------------------------- ### Get MonoVertex Status Source: https://github.com/numaproj/numaflow/blob/main/docs/quick-start.md Check the status of deployed MonoVertices. 'mvtx' is a short name for the monovertex resource. ```shell kubectl get monovertex # or "mvtx" as a short name ``` -------------------------------- ### Get Pipeline Objects with kubectl Source: https://github.com/numaproj/numaflow/blob/main/docs/core-concepts/pipeline.md Use kubectl to query Pipeline objects. 'pl' is a short name for 'pipeline'. ```sh kubectl get pipeline # or "pl" as a short name ``` -------------------------------- ### Disable Autoscaling for Pipeline Vertex Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/autoscaling.md Example of disabling autoscaling for a specific vertex within a Pipeline by setting `disabled: true`. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-vertex scale: disabled: true ``` -------------------------------- ### Example Prometheus URL Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/UI/metrics-tab.md Specifies the Prometheus service endpoint for the metrics proxy. This URL is required for the Metrics Tab to function. ```yaml http://my-release-kube-prometheus-prometheus.default.svc.cluster.local:9090 ``` -------------------------------- ### Describe Kafka Topic Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/sources/kafka.md Use this command to view details about a Kafka topic, including its partitions and leader information. ```shell #!/bin/bash kafka_2.13-3.6.1/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic quickstart-events ``` -------------------------------- ### Prevent Pipeline InterStepBufferServiceName Change Source: https://github.com/numaproj/numaflow/blob/main/docs/operations/validating-webhook.md This example shows an error when trying to update the 'interStepBufferServiceName' of a Pipeline, which is prevented by the validating webhook. ```shell Error from server (BadRequest): error when applying patch: {"metadata":{"annotations":{"kubectl.kubernetes.io/last-applied-configuration":"{\"apiVersion\":\"numaflow.numaproj.io/v1alpha1\",\"kind\":\"Pipeline\",\"metadata\":{\"annotations\":{},\"name\":\"simple-pipeline\",\"namespace\":\"numaflow-system\"},\"spec\":{\"edges\":[{\"from\":\"in\",\"to\":\"cat\"},{\"from\":\"cat\",\"to\":\"out\"}],\"interStepBufferServiceName\":\"change\",\"vertices\":[{\"name\":\"in\",\"source\":{\"generator\":{\"duration\":\"1s\",\"rpu\":5}}},{\"name\":\"cat\",\"udf\":{\"builtin\":{\"name\":\"cat\"}}},{\"name\":\"out\",\"sink\":{\"log\":{}}}]}}\\n"}},"spec":{\"interStepBufferServiceName\":\"change\",\"vertices\":[{\"name\":\"in\",\"source\":{\"generator\":{\"duration\":\"1s\",\"rpu\":5}}},{\"name\":\"cat\",\"udf\":{\"builtin\":{\"name\":\"cat\"}}},{\"name\":\"out\",\"sink\":{\"log\":{}}}]}} to: Resource: "numaflow.numaproj.io/v1alpha1, Resource=pipelines", GroupVersionKind: "numaflow.numaproj.io/v1alpha1, Kind=Pipeline" Name: "simple-pipeline", Namespace: "numaflow-system" for: "examples/1-simple-pipeline.yaml": error when patching "examples/1-simple-pipeline.yaml": admission webhook "webhook.numaflow.numaproj.io" denied the request: Cannot update pipeline with different interStepBufferServiceName ``` -------------------------------- ### Configure Fixed Window Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/user-defined-functions/reduce/windowing/fixed.md Use the `fixed` configuration under the `window` section to enable fixed windowing. Specify the `length` parameter to define the window size. ```yaml vertices: - name: my-udf udf: groupBy: window: fixed: length: duration ``` -------------------------------- ### InterStepBufferService with JetStream anti-affinity Source: https://github.com/numaproj/numaflow/blob/main/docs/core-concepts/inter-step-buffer-service.md Example of configuring pod anti-affinity for a JetStream InterStepBufferService to distribute pods across different zones. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: InterStepBufferService metadata: name: default spec: jetstream: version: latest affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - podAffinityTerm: labelSelector: matchLabels: app.kubernetes.io/component: isbsvc numaflow.numaproj.io/isbsvc-name: default topologyKey: topology.kubernetes.io/zone weight: 100 ``` -------------------------------- ### Probe Configuration Source: https://github.com/numaproj/numaflow/blob/main/docs/APIs.md Configuration for Readiness and Liveness probes, including timing and threshold settings. ```APIDOC ## Probe Configuration ### Description Configuration for Readiness and Liveness probes, including timing and threshold settings. ### Fields #### `initialDelaySeconds` (int32) * Description: Number of seconds after the container has started before liveness probes are initiated. * Optional #### `timeoutSeconds` (int32) * Description: Number of seconds after which the probe times out. * Optional #### `periodSeconds` (int32) * Description: How often (in seconds) to perform the probe. * Optional #### `successThreshold` (int32) * Description: Minimum consecutive successes for the probe to be considered successful after having failed. Defaults to 1. Must be 1 for liveness and startup. Minimum value is 1. * Optional #### `failureThreshold` (int32) * Description: Minimum consecutive failures for the probe to be considered failed after having succeeded. Defaults to 3. Minimum value is 1. * Optional ``` -------------------------------- ### Customize Pipeline Vertex Probes Source: https://github.com/numaproj/numaflow/blob/main/docs/user-guide/reference/configuration/liveness-and-readiness.md Configure liveness and readiness probes for 'numa' containers and user-defined sources within a Pipeline's vertices. Similar configurations apply to other container types like UDF, UDSink, etc. ```yaml apiVersion: numaflow.numaproj.io/v1alpha1 kind: Pipeline metadata: name: my-pipeline spec: vertices: - name: my-source containerTemplate: # For "numa" container readinessProbe: initialDelaySeconds: 30 periodSeconds: 60 livenessProbe: initialDelaySeconds: 60 periodSeconds: 120 volumes: - name: my-udsource-config configMap: name: udsource-config source: udsource: container: image: my-source:latest volumeMounts: - mountPath: /path/to/my-source-config name: my-udsource-config # For User-Defined source livenessProbe: initialDelaySeconds: 40 failureThreshold: 5 - name: my-udf containerTemplate: # For "numa" container readinessProbe: initialDelaySeconds: 20 periodSeconds: 60 livenessProbe: initialDelaySeconds: 180 periodSeconds: 60 timeoutSeconds: 50 volumes: - name: my-udf-config configMap: name: udf-config udf: container: image: my-function:latest volumeMounts: - mountPath: /path/to/my-function-config name: my-udf-config # For "udf" livenessProbe: initialDelaySeconds: 40 failureThreshold: 5 ``` -------------------------------- ### Get Numaflow Pipelines Source: https://github.com/numaproj/numaflow/blob/main/docs/quick-start.md Retrieves a list of all deployed Numaflow pipelines in the cluster. Use 'pl' as a shorthand alias for 'pipeline'. ```shell kubectl get pipeline # or "pl" as a short name ```