AWS DMS Потеря записей с использованием DatabaseMigrationService.Client.describe_table_statistics с большим набором результатов

Question

Я использую description_table_statistics для получения списка таблиц в заданной задаче DMS и условно циклически описываю description_table_statistics с ответом ['Marker'].

Когда я не использую фильтры, я получаю правильныйколичество записей, 13к +.Когда я использую фильтр или комбинацию фильтров, у которых результирующий набор меньше, чем MaxRecords, я получаю правильное количество записей.

Однако, когда я передаю фильтр, который получит набор записей больше, чем MaxRecords, я получаю гораздо меньше записей, чем должен был.

Вот моя функция для получения набора таблиц:

def get_dms_task_tables(account, region, task_name, schema_name=None, table_state=None):
   tables=[]
   max_records=500

   filters=[]
   if schema_name:
      filters.append({'Name':'schema-name', 'Values':[schema_name]})
   if table_state:
      filters.append({'Name':'table-state', 'Values':[table_state]})

   task_arn = get_dms_task_arn(account, region, task_name)

   session = boto3.Session(profile_name=account, region_name=region)
   client = session.client('dms')

   response = client.describe_table_statistics(
      ReplicationTaskArn=task_arn
      ,Filters=filters
      ,MaxRecords=max_records)

   tables += response['TableStatistics']

   while len(response['TableStatistics']) == max_records:
      response = client.describe_table_statistics(
         ReplicationTaskArn=task_arn
         ,Filters=filters
         ,MaxRecords=max_records
         ,Marker=response['Marker'])

      tables += response['TableStatistics']

   return tables

Для устранения неполадок я зацикливаюсь на таблицах, печатающих по одной строке на таблицу:

        print(', '.join((
            t['SchemaName']
            ,t['TableName']
            ,t['TableState'])))

Когда я пропускаю фильтры и grep для этого состояния таблицы «Таблица заполнена», я получаю 12k +записей, что является правильным счетом, через консоль

Так что, по крайней мере, поверхностно, цикл ответа работает.

Когда я передаю имя схемы и условия фильтра состояния таблицы, я получаюправильный счет, подтвержденный консолью, но этот счет меньше, чем MaxRecords.

Когда я просто передаю фильтр состояния таблицы для «Таблица заполнена», я получаю только 949 записей, поэтому мне не хватает 11krecords.

Я попытался опустить параметр Filter в description_table_statistics внутри цикла, но я получаю одинаковые результаты во всех случаях.

Я подозреваю, что что-то не так с моим вызовом description_table_statistics внутри цикла, но я не смог найти примеры этого в документации amazon, чтобы подтвердить это.

Answer 1

Когда применяются фильтры, description_table_statistics не соответствует пределу MaxRecords.

На самом деле, кажется, что он извлекает (2 x MaxRecords), применяет фильтр и возвращает этот набор. Или, возможно, он извлекает MaxRecords, применяет фильтр и продолжается до тех пор, пока набор результатов не станет больше, чем MaxRecords. В любом случае, мое состояние было проблемой.

Я заменил

while len(response['TableStatistics']) == max_records:

с

while 'Marker' in response:

и теперь функция возвращает правильное количество записей.

Кстати, моя первая попытка была

while len(response['TableStatistics']) >= 1:

но на последней итерации цикла он выдал эту ошибку:

KeyError: 'Marker'

Завершенная и работающая функция теперь выглядит так:

def get_dms_task_tables(account, region, task_name, schema_name=None, table_state=None):
   tables=[]
   max_records=500

   filters=[]
   if schema_name:
      filters.append({'Name':'schema-name', 'Values':[schema_name]})
   if table_state:
      filters.append({'Name':'table-state', 'Values':[table_state]})

   task_arn = get_dms_task_arn(account, region, task_name)

   session = boto3.Session(profile_name=account, region_name=region)
   client = session.client('dms')

   response = client.describe_table_statistics(
      ReplicationTaskArn=task_arn
      ,Filters=filters
      ,MaxRecords=max_records)

   tables += response['TableStatistics']

   while 'Marker' in response:
      response = client.describe_table_statistics(
         ReplicationTaskArn=task_arn
         ,Filters=filters
         ,MaxRecords=max_records
         ,Marker=response['Marker'])

      tables += response['TableStatistics']

   return tables