And another question is, could I set up different retention policy for depending which records? e.g. I want to get rid of records by using similar lookup query as above, records for users in core_free_tier should be gone sooner. Not sure if this is done by sending a task to do so periodically

Hi Julian, I don't know the answer to your ingestion-time lookup-based filtering (... although from the streaming setup wizard it looks like filters operate post-transform) For your question on retention of specific records, you are correct that you would need to reindex the data per time interval to remove the unwanted records. This is a general technique to update or delete existing data in Druid. This can be done fairly easily with an MSQ or native reindex job ... if you have any specific questions about this or want us to sanity a proposed job to do this, just post the spec or query and we can take a look. Thanks. John

Hi Julian, I don’t know the answer to your ingestion-time lookup-based filtering (... although from the streaming setup wizard it looks like filters operate post-transform)

where can I read they operate post-transform? what happens then if there is no transform? On the second topic, would I need to re-index from a file? I am looking into how I could automate this on a monthly basis for example, where I do not really depend on index_parallel from file or s3.

Ah, here it is: https://druid.apache.org/docs/latest/ingestion/ingestion-spec#filter

Note that within a

transformSpec

, the

transforms

are applied before the

filter

, so the filter can refer to a transform.

So I assume if there is no transforms then filter will get applied anyway

Yes, that's how I understand it.

If I were to run an index_parallel job, something like this would do the trick?

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "my_datasource",
      "timestampSpec": {
        "column": "ts",
        "format": "millis",
        "missingValue": null
      },
      "dimensionsSpec": {
        "dimensions": [
          {
            "type": "string",
            "name": "user_id",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "status_code",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "long",
            "name": "response_size",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": false
          },
          {
            "type": "long",
            "name": "isSubRequest",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": false
          }
        ],
        "dimensionExclusions": [
          "__time",
          "sum_units",
          "units",
          "count",
          "ts",
          "sum_response_size"
        ],
        "includeAllDimensions": false,
        "useSchemaDiscovery": false
      },
      "metricsSpec": [
        {
          "type": "count",
          "name": "count"
        },
        {
          "type": "longSum",
          "name": "sum_response_size",
          "fieldName": "response_size"
        },
        {
          "type": "longSum",
          "name": "sum_units",
          "fieldName": "units"
        }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "MINUTE",
        "rollup": true,
        "intervals": []
      },
      "transformSpec": {
        "filter": null,
        "transforms": [
          {
            "type": "expression",
            "name": "isSubRequest",
            "expression": "if(\"batch_id\" > 0, 1, 0)"
          }
        ]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "druid",
        "dataSource": "my_datasource",
        "interval": "2024-10-12/2024-11-11",
        "filter": {
          "type": "not",
          "field": {
            "type": "extraction",
            "dimension": "user_id",
            "value": "core_free_tier",
            "extractionFn": {
              "type": "registeredLookup",
              "lookup": "customer_metadata",
              "retainMissingValue": true
            }
          }
        }
      },
      "appendToExisting": false,
      "dropExisting": true
    },
    "tuningConfig": {
      "type": "index_parallel",
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 1000000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxRowsPerSegment": 5000000,
      "intermediatePersistPeriod": "PT5M",
      "maxPendingPersists": 0,
      "indexSpec": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "indexSpecForIntermediatePersists": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "reportParseExceptions": false,
      "handoffConditionTimeout": 0,
      "resetOffsetAutomatically": false,
      "skipSequenceNumberAvailabilityCheck": true,
      "chatRetries": 8,
      "httpTimeout": "PT10S",
      "shutdownTimeout": "PT80S",
      "recordBufferSize": 10000,
      "recordBufferOfferTimeout": 5000,
      "recordBufferFullWait": 5000,
      "logParseExceptions": true,
      "maxParseExceptions": 2147483647,
      "maxSavedParseExceptions": 0,
      "maxRecordsPerPoll": 100,
      "intermediateHandoffPeriod": "P2147483647D",
      "repartitionTransitionDuration": "PT120S",
      "offsetFetchPeriod": "PT30S",
      "useListShards": true,
      "skipIgnorableShards": false,
      "maxNumConcurrentSubTasks": 150
    }
  },
  "context": {
    "forceTimeChunkLock": true,
    "useLineageBasedSegmentAllocation": true,
    "priority": 80
  }
}

Also the key thing is that it checks a lookup table that has a map of user_ids with its tier as

Lookup : customer_metadata
------------------------------
key          | value
------------------------------
01234abcd    | developer_tier
098765ejf    | core_free_tier
djck1234k    | growth_tier
oldo4958d    | core_free_tier

Generally that looks right to me. There are a lot of detailed properties in that spec that I usually don't specify. My suggestion is to first run this but output to temp datasource so you can check it's results without destroying your current production data. Then once you verify it is producing what you want, change it to reindex your main datasource in place.

Yeah, we usually run that kind of tests in a dev cluster first that has no production data or usage.