老师你好，我这边按照官网输入后，在es里无法查询到自己上报数据

val env = StreamExecutionEnvironment.getExecutionEnvironment

    val input = env.socketTextStream("localhost",9999)


    val httpHosts = new java.util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("127.0.0.1",9200, "http"))

    val esSinkBuilder = new ElasticsearchSink.Builder[String](
      httpHosts,
      new ElasticsearchSinkFunction[String] {
        def createIndexRequest(element: String): IndexRequest = {
          val json = new java.util.HashMap[String, String]
          json.put("data", element)

          return Requests.indexRequest()
            .index("my-index")
            .`type`("my-type")
            .source(json)
        }

        override def process(element: String, ctx: RuntimeContext, indexer: RequestIndexer): Unit = {
          indexer.add(createIndexRequest(element))
        }
      }
    )

    input.addSink(esSinkBuilder.build)

    env.execute("SinkToElasticSearch")

我在es的查询语句为：

curl 'localhost:9200/my-index/my-type/_search?pretty'