ApsaraMQ for Kafka:Create an Elasticsearch sink connector

ApsaraMQ for Kafka

• Örneğiniz için connector özelliğini etkinleştirin.

• Veri kaynağı olarak kullanılacak bir konu oluşturun.

Function Compute

• Function Compute hizmetini etkinleştirin.

Elasticsearch

• Elasticsearch kümesi ve dizini oluşturun (Elasticsearch konsolu üzerinden). Function Compute istemcisiyle (sürüm 7.7.0) uyumluluk sağlamak için sürüm 7.0 veya sonraki bir sürümü kullanın.

• Function Compute uç noktasının CIDR bloğunu Elasticsearch izin listesine ekleyin. İlk testler sırasında VPC'deki tüm IP adreslerine izin vermek için 0.0.0.0/0 değerini belirtin; bağlantıyı doğruladıktan sonra aralığı kısıtlayın.

Toplanması gereken bilgiler

Sihirbazı başlatmadan önce aşağıdaki bilgileri toplayın.

Adım 1: Temel bilgileri yapılandırma

Configure Basic Information adımında connector adını belirleyin ve örnek ayrıntılarını gözden geçirin.

Next seçeneğine tıklayın.

Adım 2: Kaynak hizmeti yapılandırma

Configure Source Service adımında kaynak hizmet olarak Message Queue for Apache Kafka seçeneğini belirleyin ve aşağıdaki parametreleri yapılandırın.

Ek parametreleri genişletmek için Configure Runtime Environment seçeneğine tıklayın.

Adım 3: Hedef hizmeti yapılandırma

Configure Destination Service adımında hedef hizmet olarak Elasticsearch seçeneğini belirleyin ve aşağıdaki parametreleri yapılandırın.

Create seçeneğine tıklayın.

Connector'ü dağıtma

Oluşturma işleminden sonra connector, Connectors sayfasında görünür. Connector'ü başlatmak için Actions sütunundaki Deploy seçeneğine tıklayın.

Test mesajı gönderme

1. Connectors sayfasında connector'ü bulun ve Actions sütunundaki Test seçeneğine tıklayın.

2. Send Message panelinde Method of Sending değerini Console olarak ayarlayın.

3. Message Key alanına bir anahtar girin; örneğin demo.

4. Message Content alanına bir JSON gövdesi girin; örneğin:

      {"key": "test"}

5. Send to Specified Partition seçeneği için belirli bir parçayı hedeflemek üzere Yes seçeneğine tıklayın ve bir Partition ID (örneğin 0) girin veya sistem tarafından atanmasını sağlamak için No seçeneğine tıklayın. Parça ID'lerini aramak için bkz. Parça durumunu görüntüleme.

Test mesajlarını Docker veya bir SDK üzerinden de gönderebilirsiniz. Method of Sending alanında ilgili seçeneği belirleyin ve ekrandaki talimatları izleyin.

Elasticsearch dizinini kontrol etme

1. Kibana konsoluna giriş yapın.

2. Hedef dizinde arama yapmak için aşağıdaki sorguyu çalıştırın:

      GET /<index_name>/_search

3. Yanıtın gönderdiğiniz mesajı içerdiğini onaylayın. Başarılı bir yanıt şu şekildedir:

      {
        "took": 8,
        "timed_out": false,
        "_shards": {
          "total": 5,
          "successful": 5,
          "skipped": 0,
          "failed": 0
        },
        "hits": {
          "total": {
            "value": 1,
            "relation": "eq"
          },
          "max_score": 1.0,
          "hits": [
            {
              "_index": "product_****",
              "_type": "_doc",
              "_id": "TX3TZHgBfHNEDGoZ****",
              "_score": 1.0,
              "_source": {
                "msg_body": {
                  "key": "test",
                  "offset": 2,
                  "overflowFlag": false,
                  "partition": 2,
                  "timestamp": 1616599282417,
                  "topic": "dv****",
                  "value": "test1",
                  "valueSize": 8
                },
                "doc_as_upsert": true
              }
            }
          ]
        }
      }