Streamsets: есть ли способ подсчета записей в теме Kafka, используя streamsets - PullRequest
Новая
       10

Streamsets: есть ли способ подсчета записей в теме Kafka, используя streamsets

1 голос
/ 12 мая 2019

Я использую StreamSets в качестве инструмента для загрузки записей из базы данных Oracle в разделы Kafka.Теперь я хочу использовать его через сам StreamSets, а также посчитать количество записей в темах Kafka.

Как я могу это сделать.Пожалуйста, помогите

1 Ответ

1 голос
/ 14 мая 2019

Вы можете использовать API REST history Сборщика данных StreamSets для получения данных с количеством записей для каждого этапа. Например, вот счетчики для последнего запуска данного конвейера. Я использую превосходный инструмент jq для анализа JSON в командной строке. 

$ curl -s -u admin:admin -H 'X-Requested-By:sdc' http://localhost:18630/rest/v1/pipeline/RedshiftStreamingwithKinesisFirehose537add73-bb16-4358-a26a-a51576dea32b/history | jq -r .[0].metrics | jq .counters
{
  "pipeline.batchCount.counter": {
    "count": 1029
  },
  "pipeline.batchErrorMessages.counter": {
    "count": 0
  },
  "pipeline.batchErrorRecords.counter": {
    "count": 0
  },
  "pipeline.batchInputRecords.counter": {
    "count": 648226
  },
  "pipeline.batchOutputRecords.counter": {
    "count": 648226
  },
  "stage.ExpressionEvaluator_01.errorRecords.counter": {
    "count": 0
  },
  "stage.ExpressionEvaluator_01.inputRecords.counter": {
    "count": 648226
  },
  "stage.ExpressionEvaluator_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.ExpressionEvaluator_01.stageErrors.counter": {
    "count": 0
  },
  "stage.ExpressionEvaluator_01:ExpressionEvaluator_01OutputLane15561338960790.outputRecords.counter": {
    "count": 648226
  },
  "stage.FieldOrder_01.errorRecords.counter": {
    "count": 0
  },
  "stage.FieldOrder_01.inputRecords.counter": {
    "count": 648226
  },
  "stage.FieldOrder_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.FieldOrder_01.stageErrors.counter": {
    "count": 0
  },
  "stage.FieldOrder_01:FieldOrder_01OutputLane15561351879260.outputRecords.counter": {
    "count": 648226
  },
  "stage.FieldTypeConverter_01.errorRecords.counter": {
    "count": 0
  },
  "stage.FieldTypeConverter_01.inputRecords.counter": {
    "count": 648226
  },
  "stage.FieldTypeConverter_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.FieldTypeConverter_01.stageErrors.counter": {
    "count": 0
  },
  "stage.FieldTypeConverter_01:FieldTypeConverter_01OutputLane15560499048280.outputRecords.counter": {
    "count": 648226
  },
  "stage.KinesisFirehose_01.errorRecords.counter": {
    "count": 0
  },
  "stage.KinesisFirehose_01.inputRecords.counter": {
    "count": 648226
  },
  "stage.KinesisFirehose_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.KinesisFirehose_01.stageErrors.counter": {
    "count": 0
  },
  "stage.MySQLBinaryLog_01.errorRecords.counter": {
    "count": 0
  },
  "stage.MySQLBinaryLog_01.inputRecords.counter": {
    "count": 0
  },
  "stage.MySQLBinaryLog_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.MySQLBinaryLog_01.stageErrors.counter": {
    "count": 0
  },
  "stage.MySQLBinaryLog_01:MySQLBinaryLog_01OutputLane15561313696850.outputRecords.counter": {
    "count": 648226
  },
  "stage.StreamSelector_01.errorRecords.counter": {
    "count": 0
  },
  "stage.StreamSelector_01.inputRecords.counter": {
    "count": 648226
  },
  "stage.StreamSelector_01.outputRecords.counter": {
    "count": 648226
  },
  "stage.StreamSelector_01.stageErrors.counter": {
    "count": 0
  },
  "stage.StreamSelector_01:StreamSelector_01OutputLane1556133811620.outputRecords.counter": {
    "count": 0
  },
  "stage.StreamSelector_01:StreamSelector_01OutputLane1556133816638.outputRecords.counter": {
    "count": 648226
  },
  "stage.Trash_01.errorRecords.counter": {
    "count": 0
  },
  "stage.Trash_01.inputRecords.counter": {
    "count": 0
  },
  "stage.Trash_01.outputRecords.counter": {
    "count": 0
  },
  "stage.Trash_01.stageErrors.counter": {
    "count": 0
  }
}
...