Настройка неупорядоченного режима для непрерывной ингестии

Пошаговое руководство по настройке неупорядоченного режима для непрерывной ингестии в ClickPipes для S3.

По умолчанию S3 ClickPipe предполагает, что файлы добавляются в бакет в лексикографическом порядке. S3 ClickPipe можно настроить на приём файлов, у которых нет подразумеваемого порядка, указав очередь Amazon SQS, подключённую к бакету, и при необходимости используя Amazon EventBridge в качестве маршрутизатора событий. Это позволяет ClickPipes отслеживать события ObjectCreated:* и принимать любые новые файлы независимо от схемы их именования.

Неупорядоченный режим поддерживается только для Amazon S3 и не поддерживается для публичных бакетов или S3-совместимых сервисов. Для него требуется настроить очередь Amazon SQS, подключённую к бакету, и при необходимости использовать Amazon EventBridge в качестве маршрутизатора событий.

​

Как это работает

В этом режиме S3 ClickPipe выполняет первоначальную загрузку всех файлов по выбранному пути, а затем отслеживает в очереди события ObjectCreated:*, соответствующие указанному пути. Любое сообщение об уже встречавшемся файле, файле, не соответствующем пути, или событии другого типа будет игнорироваться. Ингестия файлов начинается при достижении порога, заданного в max insert bytes или max file count, либо по истечении настраиваемого интервала (по умолчанию 30 секунд).

Чтобы принимать только файлы, доставленные через очередь, включите Пропустить первоначальную загрузку. Когда этот параметр включён, ClickPipes пропускает первоначальное сканирование существующих файлов по выбранному пути и обрабатывает только файлы, доставленные через очередь SQS.

При ингестии данных возможны различные сбои, которые могут приводить к частичной вставке или дублированию данных. ClickPipes для объектного хранилища устойчивы к сбоям вставки и обеспечивают семантику «ровно один раз» с помощью временных staging-таблиц. Сначала данные вставляются в staging-таблицу; если что-то пойдёт не так, staging-таблица очищается, и вставка повторяется из чистого состояния. Только после успешного завершения вставки партиции перемещаются в целевую таблицу.

1. Создание очереди Amazon SQS

   1. В консоли AWS перейдите в раздел Simple Queue Service > Create queue. Используйте настройки по умолчанию для создания новой стандартной очереди.

   Одна очередь на каждый ClickPipe

   Для каждого ClickPipe нужна отдельная выделенная очередь SQS. SQS доставляет каждое сообщение только одному consumer: если одну и ту же очередь используют два или более ClickPipe (или любое другое приложение), каждое событие получает только один из них, а пайпы, которые это событие не получили, молча пропускают соответствующие файлы. Это верно даже в том случае, если ClickPipes используют разные шаблоны пути: пайп, получивший событие для файла, который не соответствует шаблону, отбрасывает его, и никакой другой пайп это событие уже не увидит. Чтобы выполнять приём данных из одного и того же бакета с помощью нескольких ClickPipes, создайте отдельную очередь для каждого ClickPipe и используйте правила EventBridge, чтобы направлять события в каждую очередь.

   Мы настоятельно рекомендуем настроить Dead-Letter-Queue (DLQ) для очереди SQS, чтобы упростить отладку и повторную обработку сообщений с ошибками. Если DLQ настроена, такие сообщения будут снова помещаться в очередь и обрабатываться повторно до числа попыток, заданного в параметре DLQ maxReceiveCount.

   2. Подключите S3 бакет к очереди SQS одним из двух способов, описанных ниже. Для большинства сценариев рекомендуется использовать EventBridge: он поддерживает fan-out, обеспечивает более гибкую фильтрацию событий и не имеет ограничения S3, допускающего только одно правило уведомления на тип события и prefix.

   через EventBridge

   a. В свойствах S3 бакета перейдите в Event notifications > Amazon EventBridge и включите отправку уведомлений в EventBridge. Нажмите Save changes.

   b. В консоли AWS перейдите в Amazon EventBridge > Rules > Create rule. Укажите имя правила (например, S3ObjectCreated), выберите шину событий default и нажмите Next. На шаге Build event pattern выберите AWS events or EventBridge partner events в качестве источника событий, затем вручную введите следующий шаблон события, заменив <bucket-name> именем вашего бакета:

   {
     "source": ["aws.s3"],
     "detail-type": ["Object Created"],
     "detail": {
       "bucket": {
         "name": ["<bucket-name>"]
       }
     }
   }
   

   При необходимости добавьте в шаблон условие object.key, чтобы фильтровать объекты по префиксу или суффиксу. Если добавите его, убедитесь, что оно соответствует пути, заданному для ClickPipe.

   c. На шаге Select target(s) выберите AWS service в качестве типа цели и укажите SQS queue. Выберите очередь, созданную на предыдущем шаге. Оставьте флажок Use execution role (recommended) установленным, чтобы EventBridge автоматически создал необходимую роль IAM, затем нажмите Next и завершите настройку.

   d. Измените политику доступа очереди SQS, чтобы разрешить EventBridge отправлять в неё сообщения. Замените <sqs-queue-arn> и <eventbridge-rule-arn> соответствующими значениями:

   {
     "Version": "2012-10-17",
     "Id": "example-ID",
     "Statement": [
       {
         "Sid": "AllowEventBridgeToSendMessage",
         "Effect": "Allow",
         "Principal": {
           "Service": "events.amazonaws.com"
         },
         "Action": "SQS:SendMessage",
         "Resource": "<sqs-queue-arn>",
         "Condition": {
           "ArnLike": {
             "aws:SourceArn": "<eventbridge-rule-arn>"
           }
         }
       }
     ]
   }
   

   Напрямую S3 → SQS

   a. Измените политику доступа очереди SQS, чтобы разрешить вашему S3 бакету отправлять в неё сообщения. Замените <sqs-queue-arn>, <bucket-arn> и <aws-account-id> соответствующими значениями:

   {
     "Version": "2012-10-17",
     "Id": "example-ID",
     "Statement": [
       {
         "Sid": "AllowS3ToSendMessage",
         "Effect": "Allow",
         "Principal": {
           "Service": "s3.amazonaws.com"
         },
         "Action": "SQS:SendMessage",
         "Resource": "<sqs-queue-arn>",
         "Condition": {
           "ArnLike": {
             "aws:SourceArn": "<bucket-arn>"
           },
           "StringEquals": {
             "aws:SourceAccount": "<aws-account-id>"
           }
         }
       }
     ]
   }
   

   b. В свойствах S3 бакета включите Event notifications, выберите All object create events (s3:ObjectCreated:*) в качестве типа события и укажите в качестве назначения очередь SQS. Не выбирайте отдельные типы событий: в зависимости от способа загрузки могут генерироваться разные события ObjectCreated (например, multipart-загрузки генерируют CompleteMultipartUpload, а не Put), и файлы, тип события которых не выбран, никогда не будут доставлены в очередь. При необходимости укажите префикс или суффикс, чтобы отфильтровать объекты, для которых будут отправляться уведомления, — если сделаете это, убедитесь, что они соответствуют пути, заданному для ClickPipe.

   S3 не позволяет использовать несколько пересекающихся правил уведомлений для одних и тех же типов событий в одном бакете. Если для событий ObjectCreated в этом бакете у вас уже настроено правило уведомлений, используйте вместо этого вариант с EventBridge.

2. Настройте роль IAM

   1. В консоли ClickHouse Cloud перейдите в Settings > Network security information и скопируйте IAM role ARN вашего сервиса.

   2. В консоли AWS перейдите в IAM > Roles > Create role. Выберите Custom trust policy и вставьте следующий текст, заменив <ch-cloud-arn> на IAM role ARN, скопированный на предыдущем шаге:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowAssumeRole",
         "Effect": "Allow",
         "Principal": {
           "AWS": "<ch-cloud-arn>"
         },
         "Action": "sts:AssumeRole"
       }
     ]
   }
   

   3. Создайте встроенную политику для роли IAM с необходимыми разрешениями, чтобы читать объекты в S3 и управлять сообщениями в очереди SQS. Замените <bucket-arn> и <sqs-queue-arn> на соответствующие значения:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "S3BucketMetadataAccess",
         "Effect": "Allow",
         "Action": [
           "s3:GetBucketLocation",
           "s3:ListBucket"
         ],
         "Resource": "<bucket-arn>"
       },
       {
         "Sid": "AllowGetListObjects",
         "Effect": "Allow",
         "Action": [
           "s3:Get*",
           "s3:List*"
         ],
         "Resource": "<bucket-arn>/*"
       },
       {
         "Sid": "SQSNotificationsAccess",
         "Effect": "Allow",
         "Action": [
           "sqs:DeleteMessage",
           "sqs:ListQueues",
           "sqs:ReceiveMessage",
           "sqs:GetQueueAttributes"
         ],
         "Resource": "<sqs-queue-arn>"
       }
     ]
   }
   
3. Создайте ClickPipe в неупорядоченном режиме

   1. В консоли ClickHouse Cloud перейдите в раздел Data Sources > Create ClickPipe и выберите Amazon S3. Введите данные для подключения к вашему S3 бакету. В разделе Authentication method выберите роль IAM и укажите ARN роли, которую вы создали на предыдущем шаге.

   2. В разделе Incoming data включите Continuous ingestion. В качестве режима ингестии выберите Any order и укажите URL очереди SQS для очереди, связанной с вашим бакетом. Чтобы обрабатывать только файлы из очереди и не выполнять первоначальное сканирование существующих файлов по выбранному пути, включите Skip initial load. Эта очередь должна использоваться исключительно данным ClickPipe — см. Создание очереди Amazon SQS.

   3. В разделе Parse information задайте ключ сортировки для целевой таблицы. При необходимости скорректируйте сопоставленную схему, затем настройте роль для пользователя базы данных ClickPipes.

   4. Проверьте конфигурацию и нажмите Create ClickPipe. Если Skip initial load не включен, ClickPipes выполнит первоначальное сканирование вашего бакета, чтобы загрузить все существующие файлы, соответствующие указанному пути, а затем начнёт обрабатывать файлы по мере доставки из очереди соответствующих событий ObjectCreated:*.