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

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

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

через EventBridgeНапрямую S3 → SQS

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>"
        }
      }
    }
  ]
}

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>"
    }
  ]
}

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:*.