https://linen.dev logo
Join Slack
Powered by
This message was deleted.
# troubleshooting
s
This message was deleted.
s
Active real-time ingestion holds a lock for time intervals that are being ingested. You're getting overlapping locks, because your load from S3 goes through 2022-09-29 which you are currently loading. If you limit the interval for your backfill up to 2022-09-28T000000, you should be able to do the load.
j
ah I see, makes sense because realtime ingestion should already have covered 
2022-09-29T00:00:00
if the realtime ingestion started on the 27th for example
👍 1
I was also wondering two things: • I started realtime ingestion, say on 
2022-09-27T16:43:33
and I start S3 index parallel to backfill data, essentially from 
2022-09-01T00:00:00
to 
2022-09-28T00:00:00
, would there be any duplicate data? or there would be segment replaced by the index-parallel? • Second question I have is about speeding up index-parallel, what settings should I tweak so that indexing 1 month worth of data from S3 takes a day or two? Taking into account that there is around 9B json records per day.
d
1. If you did not turn on append mode, then the last one will win and overwrite the existing time bucketed data. 2. If it’s dynamic partition, you can just spawn more subtasks, but later you need a compaction job to optimize the file size so that they are not too small
j
in case you can see something weird, the spec looks like:
Copy code
{
  "type": "index_parallel",
  "id": "index_parallel_analytics_heodejnl_2022-09-29T15:07:01.433Z",
  "groupId": "index_parallel_analytics_heodejnl_2022-09-29T15:07:01.433Z",
  "resource": {
    "availabilityGroup": "index_parallel_analytics_heodejnl_2022-09-29T15:07:01.433Z",
    "requiredCapacity": 1
  },
  "spec": {
    "dataSchema": {
      "dataSource": "analytics",
      "timestampSpec": {
        "column": "ts",
        "format": "millis",
        "missingValue": null
      },
      "dimensionsSpec": {
        "dimensions": [
          {
            "type": "string",
            "name": "user_id",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "ip",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "proxy",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "string",
            "name": "status_code",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": true
          },
          {
            "type": "long",
            "name": "payload_size",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": false
          },
          {
            "type": "long",
            "name": "isAuthed",
            "multiValueHandling": "SORTED_ARRAY",
            "createBitmapIndex": false
          }
        ],
        "dimensionExclusions": [
          "__time",
          "count",
          "ts"
        ],
        "includeAllDimensions": false
      },
      "metricsSpec": [
        {
          "type": "count",
          "name": "count"
        },
        {
          "type": "longSum",
          "name": "sum_payload_size",
          "fieldName": "payload_size"
        }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "MINUTE",
        "rollup": true,
        "intervals": [
          "2022-09-01T00:00:00.000Z/2022-09-28T00:00:00.000Z"
        ]
      },
      "transformSpec": {
        "filter": null,
        "transforms": [
          {
            "type": "expression",
            "name": "isAuthed",
            "expression": "if(\"auth_id\" > 0, 1, 0)"
          }
        ]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "uris": null,
        "prefixes": [
          "<s3://analytics-cold-storage-v2/2022/09/>"
        ],
        "objects": null,
        "properties": null,
        "proxyConfig": null,
        "endpointConfig": null,
        "clientConfig": null
      },
      "inputFormat": {
        "type": "json",
        "flattenSpec": {
          "useFieldDiscovery": true,
          "fields": []
        },
        "featureSpec": {},
        "keepNullColumns": true
      },
      "appendToExisting": false,
      "dropExisting": false
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsPerSegment": 5000000,
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 1000000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxTotalRows": null,
      "numShards": null,
      "splitHintSpec": null,
      "partitionsSpec": {
        "type": "dynamic",
        "maxRowsPerSegment": 5000000,
        "maxTotalRows": null
      },
      "indexSpec": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "metricCompression": "lz4",
        "longEncoding": "longs",
        "segmentLoader": null
      },
      "indexSpecForIntermediatePersists": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "metricCompression": "lz4",
        "longEncoding": "longs",
        "segmentLoader": null
      },
      "maxPendingPersists": 0,
      "forceGuaranteedRollup": false,
      "reportParseExceptions": false,
      "pushTimeout": 0,
      "segmentWriteOutMediumFactory": null,
      "maxNumConcurrentSubTasks": 5,
      "maxRetry": 3,
      "taskStatusCheckPeriodMs": 1000,
      "chatHandlerTimeout": "PT10S",
      "chatHandlerNumRetries": 5,
      "maxNumSegmentsToMerge": 100,
      "totalNumMergeTasks": 10,
      "logParseExceptions": true,
      "maxParseExceptions": 2147483647,
      "maxSavedParseExceptions": 0,
      "maxColumnsToMerge": -1,
      "awaitSegmentAvailabilityTimeoutMillis": 0,
      "maxAllowedLockCount": -1,
      "partitionDimensions": []
    }
  },
  "context": {
    "forceTimeChunkLock": true,
    "useLineageBasedSegmentAllocation": true
  },
  "dataSource": "analytics"
}
the datasource is fully compacted too
d
Copy code
"appendToExisting": false,
 "dropExisting": false
ok, then the last one will win and overwrite the older data.
Copy code
"maxNumConcurrentSubTasks": 5
This is too small, jack it up to 100 or 200
but then each of your zip file will be so small. Therefore, you need a compaction job. But data will be available very quickly.
j
would not that eat too much memory if there is kinesis ingestion also happening?
d
Yes, you need to do that memory calculation very carefully.
we separated our Kafka MM vs batch MM, they don’t bulldoze each other.
j
given that this will probably be only one shot to backfill, I do not think I need to split them
99% of time will be kinesis ingestion
just to double check, what parameters I need to check to calculate the memory? and also, is it possible to increase 
maxNumConcurrentSubTasks
while already running for couple of hours?
d
you have to try ingesting again, you cannot modify currently running ingestion.
This is the most basic settings that’s relevant to you.
Copy code
# Indexer (worker/peon) Java options
# Make sure -Xmx is large enough for DRUIDMM_INDEXER_TASK_DEFAULTROWFLUSHBOUNDARY but not so big that we lose precious pods capacity.
# With our largest dataset, the -Xmx cannot be smaller than 19g.
DRUIDMM_INDEXER_RUNNER_JAVAOPTS: -server -Xms128m -Xmx19g -Duser.timezone=UTC -Dfile.encoding=UTF-8 -XX:+UseG1GC -XX:+UseStringDeduplication -XX:+ExitOnOutOfMemoryError -Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager -Daws.region=us-west-2 --add-exports=jdk.management/com.sun.management.internal=ALL-UNNAMED --add-exports=java.base/jdk.internal.perf=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED

DRUIDMM_INDEXER_TASK_DEFAULTROWFLUSHBOUNDARY: "100000" # For ballpark comparison, each LARGEST_TABLE's row is about 2kb. So about 200mb is buffered inside peon indexer.

DRUIDMM_DAEMON_MAX_HEAP_SIZE: 36g    # -20% of max memory. E.g. 45GB - (20% x 45GB) = 36GB, each thread will get 18GB heap size

# processing
DRUIDMM_WORKER_CAPACITY: "2"
DRUIDMM_PROCESSING_BUFFER_SIZEBYTES: "100000000"
DRUIDMM_PROCESSING_NUMMERGEBUFFERS: "2"
DRUIDMM_PROCESSING_NUMTHREADS: "2"
DRUIDMM_PROCESSING_TMPDIR: var/ebs/middlemanager/processing
DRUIDMM_PROCESSING_STORAGE_TYPE: local

# Kubernetes
resources:
limits:
  # There are 2 workers. See: DRUIDMM_WORKER_CAPACITY
  cpu: 2
  # Each worker can consume 19GB RAM: 2 * 19 = 38.
  # See: -Xmx setting inside DRUIDMM_INDEXER_RUNNER_JAVAOPTS.
  # Now we need to add a little bit more RAM because the -Xmx should only be 80% of total RAM.
  # So the math is: (2 * 19) / 4 * 5 = 47.5
  memory: 47Gi
requests:
  cpu: 2
  memory: 47Gi
j
all right, I checked 
/druid/indexer/v1/workers
to find out capacity and there are few slots available really, from 12 MM, most of them already with up to 10 slots taken out of 12 slots available each
my MM looks smaller I think
Copy code
middleManager: {
    replicaCount: 12,
    config: {
        DRUID_XMX: '5g',
        DRUID_XMS: '5g',
        DRUID_MAXDIRECTMEMORYSIZE: '10g',
        druid_node_type: 'middleManager',
        druid_indexer_fork_property_druid_processing_buffer_sizeBytes: '300000000',
        druid_worker_capacity: '12',
        druid_service: 'druid/middleManager',
        druid_indexer_task_baseDir: '/opt/druid/var/tmp',
        druid_indexer_fork_property_druid_processing_numMergeBuffers: '2',
        druid_indexer_fork_property_druid_processing_numThreads: '2',
        druid_indexer_task_restoreTasksOnRestart: 'true',
        druid_indexer_task_gracefulShutdownTimeout: 'PT120S',
        druid_indexer_runner_javaOpts: '-server -Xmx5g -XX\:+IgnoreUnrecognizedVMOptions -XX\:MaxDirectMemorySize\=10g -Duser.timezone\=UTC -XX\:+PrintGC -XX\:+PrintGCDateStamps -XX\:+ExitOnOutOfMemoryError -XX\:+HeapDumpOnOutOfMemoryError -XX\:HeapDumpPath\=/mnt/tmp/druid-peon.hprof -Dfile.encoding\=UTF-8 -Djava.util.logging.manager\=org.apache.logging.log4j.jul.LogManager',
        druid_indexer_fork_property_druid_server_http_numThreads: '50'
    },
    persistence: {
        size: '50Gi'
    }
},
wonder if I could get better S3 performance by increasing DRUID_XMX to 19g too and DRUID_MAXDIRECTMEMORYSIZE to 36g. Wondering tho where did you get 47Gi for memory for k8s limits, in my case I currently have
Copy code
requests: {
    memory: '5g'
},
limits: {
    memory: '5g'
}
you did 
(2 * 19) / 4 * 5 = 47.5
but not sure where the 
4
and 
5
came from
d
That comes from experience, JVM xmx setting is a soft limit, in reality JVM can use even more than what the xmx settings, so I give it 20% extra padding
2 Views
Open in Slack
PreviousNext
Powered by