数据湖探索 DLI-从Kafka读取数据写入到Elasticsearch:步骤6：发送数据和查询结果

时间：2024-11-16 13:16:39

数据湖探索 DLI

步骤6：发送数据和查询结果

Kafaka端发送数据。

使用Kafka客户端向步骤2：创建Kafka的Topic中的Topic发送数据，模拟实时数据流。

发送样例数据如下：

{"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}

{"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0002", "user_name":"Jason", "area_id":"330106"}

查看Elasticsearch端数据处理后的相应结果。

发送成功后，在 CSS 集群的Kibana中执行下述语句并查看相应结果：

GET shoporders/_search

查询结果返回如下：

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "shoporders",
        "_type" : "_doc",
        "_id" : "6fswzIAByVjqg3_qAyM1",
        "_score" : 1.0,
        "_source" : {
          "order_id" : "202103241000000001",
          "order_channel" : "webShop",
          "order_time" : "2021-03-24 10:00:00",
          "pay_amount" : 100.0,
          "real_pay" : 100.0,
          "pay_time" : "2021-03-24 10:02:03",
          "user_id" : "0001",
          "user_name" : "Alice",
          "area_id" : "330106"
        }
      },
      {
        "_index" : "shoporders",
        "_type" : "_doc",
        "_id" : "6vs1zIAByVjqg3_qyyPp",
        "_score" : 1.0,
        "_source" : {
          "order_id" : "202103241606060001",
          "order_channel" : "appShop",
          "order_time" : "2021-03-24 16:06:06",
          "pay_amount" : 200.0,
          "real_pay" : 180.0,
          "pay_time" : "2021-03-24 16:10:06",
          "user_id" : "0002",
          "user_name" : "Jason",
          "area_id" : "330106"
        }
      }
    ]
  }
}

上一篇：数据湖探索 DLI-获取Flink作业委托临时凭证用于访问其他云服务:约束限制

下一篇：数据湖探索 DLI-从Kafka读取数据写入到Elasticsearch:场景描述