Bento Cookbooks

# Bento pipeline: enriched-good  ->  enriched-good-json
input:
  kafka_franz:
    seed_brokers: ["warp:9092"]
    topics: ["enriched-good"]
    consumer_group: "pipeline:enriched-good-json"

pipeline:
  processors:
    - opensnowcat:
        output_format: json
                         
output:
  kafka_franz:
    seed_brokers: ["warp:9092"]
    topic: "enriched-good-json"
    batching:
      count: 10

# Bento pipeline: enriched-good  ->  elasticsearch
input:
  kafka_franz:
    seed_brokers: ["warp:9092"]
    topics: ["enriched-good"]
    consumer_group: "pipeline:enriched-good:elastic"

pipeline:
  processors:
    - opensnowcat:
        output_format: json
      
output:
# See docs for config options https://warpstreamlabs.github.io/bento/docs/components/outputs/elasticsearch_v2
  elasticsearch_v2:
      urls: [""] #
      basic_auth:
        enabled: true
        username: "" # No default (required)
        password: "" # No default (required)
      index: "good" # No default (required)
      id: ${!count("elastic_ids")}-${!timestamp_unix()}
      discover_nodes_on_start: false
      discover_nodes_interval: 0s
      max_in_flight: 64
      batching:
        count: 5
      retry_on_status:
        - 502
        - 503
        - 504

# Bento pipeline: PII & IP Anonymization & Bot Filtering
# Bento pipeline: enriched-good  ->  enriched-good-anonymized
input:
  kafka_franz:
    seed_brokers: ["warp:9092"]
    topics: ["enriched-good"]
    consumer_group: "pipeline:enriched-good-json"

pipeline:
  processors:
    - opensnowcat:
        output_format: json
        filters:
          drop:
            nl.basjes.yauaa_context.agentClass: 
              contains: ["Bot"]
          transform:
            salt: "your-secret-salt-here"
            hash_algo: SHA-256
            fields:
              user_id:
                strategy: hash
              user_ipaddress:
                strategy: anonymize_ip
                anon_octets: 2
                anon_segments: 3
              network_userid:
                strategy: redact
                redact_value: "[REDACTED]"   
                         
output:
  kafka_franz:
    seed_brokers: ["warp:9092"]
    topic: "enriched-good-anonymized"
    batching:
      count: 10