Apache StreamPipes Documentation

The Zusammenfügen (Join) processor merges two event streams by combining their properties. It is designed for real-time stream merging, stateful event processing, and creating unified event views from multiple sources.

Key Capabilities

• Real-time Merging: Merges streams as events arrive.
• State Tracking: Maintains the last known state of the last event from each stream.
• Custom Field Selection: Allows you to specify exactly which fields from each input stream should be included in the output event.
• Dynamic Composition: Builds composite event structures dynamically.

Requirements

The processor requires two input streams:

1. First Stream: Any event stream containing at least one property.
2. Second Stream: Any event stream containing at least one property.

How it Works

1. The processor keeps the last event from each stream in memory.
2. When a new event arrives on either stream, the processor merges the selected fields from the new event with the fields from the last known event of the other stream.
3. The resulting combined event is forwarded to the next stage of the pipeline.

Note: The internal state (the last event from each stream) is cleared when the pipeline is stopped.

#### Input Stream 1
```json
{
  "deviceId": "sensor01",
  "temperature": 25.5,
  "timestamp": 1586380104915
}

Input Stream 2

{
  "location": "room1",
  "humidity": 45.2,
  "timestamp": 1586380104915
}

Configuration

• Ausgabefelder (Output Fields): deviceId, temperature, location, humidity

Output Event

{
  "deviceId": "sensor01",
  "temperature": 25.5,
  "location": "room1",
  "humidity": 45.2
}

Apache StreamPipes is an Industrial IoT data platform composed of multiple services. Its security model is based on a multi-service deployment where certain components are considered trusted dependencies within an operator-controlled perimeter.

Core Components and Security Surfaces

• streampipes-rest: The primary HTTP/REST control-plane boundary. It handles authentication, user/resource management, pipeline control, adapter control, and data-lake queries. It is not the external data-ingress path.
• Connect source/input adapters (extensions): The external data-ingress path. These connect to external sources (MQTT, Kafka, etc.), normalize raw data into events, and publish them to the internal bus.
• Pipeline elements (processors/sinks): Logic that runs over event streams; these are deployable as extensions.
• Internal message bus: The internal event transport. It is not directly writable from outside the deployment perimeter.
• Data lake: Provides time-series persistence and queryable event history.
• UI (ui/): A browser client that interacts with the control plane via the REST layer.

Trust Boundaries

1. Control Boundary: Requests from the UI/API to the StreamPipes control plane. These are authenticated and authorized via a two-tier system (role/privilege gate + per-object ACL).
2. Data-Ingress Boundary: Data flowing from external industrial sources through source/input adapters. While StreamPipes ensures parser safety for built-in paths, the semantic trustworthiness of the data is the operator's responsibility.

The Projection processor allows you to select a subset of fields from incoming events and create a new event containing only those specified fields. This is primarily used for reducing data volume, focusing on relevant data, ensuring data privacy by removing sensitive fields, and optimizing stream processing efficiency.

How it works

1. Input: Any event stream containing one or more fields.
2. Configuration: During pipeline development, you specify the list of fields you want to keep.
3. Output: A new event containing only the selected fields. All non-selected fields are completely removed from the output.

Key Characteristics

• Value Preservation: The processor retains the original values of the selected fields.
• Field Order: The order of fields in the output event may differ from the input event.
• Type Support: All field types are supported.
• Scalability: The processor can handle an arbitrary number of fields.

#### Eingabe-Ereignis
```json
{
  "temperature": 25.5,
  "humidity": 60,
  "pressure": 1013,
  "timestamp": 1586380104915,
  "device_id": "sensor_001",
  "location": "room_101"
}

Konfiguration

Ausgewählte Felder:

• temperature
• humidity
• timestamp

Ausgabe-Ereignis

{
  "temperature": 25.5,
  "humidity": 60,
  "timestamp": 1586380104915
}

The Field Hasher processor is a data transformation component that applies cryptographic hash functions to string values within a data stream. It is used to encode sensitive information, generate unique identifiers, or transform data for privacy and security purposes.

Requirements

• The input event stream must contain at least one field of type String to be used as input for the hash function.

Configuration Options

• Feld (Field): Specify the exact name of the string field to be encoded. This field must exist in the input event stream and contain string values.
• Hash-Algorithmus (Hash Algorithm): Select the algorithm used for encoding. Available options are:
  • SHA1: Produces a 40-character hexadecimal hash.
  • SHA2: Produces a 64-character hexadecimal hash (SHA-256 implementation).
  • MD5: Produces a 32-character hexadecimal hash.

Output Behavior

The processor modifies the input event by replacing the value of the selected field with its hashed version. All other fields in the event remain unchanged.

// Input Event Example
{
  "timestamp": 1617183834000,
  "sensorId": "sensor123",
  "value": 42.5,
  "user": "john.doe@example.com"
}

// Configuration
// Feld: user
// Hash-Algorithmus: MD5

// Output Event Example
{
  "timestamp": 1617183834000,
  "sensorId": "sensor123",
  "value": 42.5,
  "user": "e87f955d3b3499b8b13e901fd61b6b64"
}