elasticsearch 使用Alpakka将文档插入Elastic

wf82jlnq 于 2022-11-02 发布在 ElasticSearch

关注(0)|答案(1)|浏览(148)

我试图学习如何使用Alpakka，并设置了一个测试，写一个文档到Elastic.从阅读文档，包括https://doc.akka.io/docs/alpakka/current/elasticsearch.html写了以下内容：

import akka.actor.ActorSystem
import akka.stream.alpakka.elasticsearch.scaladsl.ElasticsearchSink
import akka.stream.alpakka.elasticsearch._
import akka.stream.scaladsl.Source
import spray.json.DefaultJsonProtocol._
import spray.json.{JsonFormat, _}

object AlpakkaWrite extends App{

  case class VolResult(symbol : String, vol : Double, timestamp : Long)

  implicit val actorSystem = ActorSystem()

  val connectionString = "****";
  val userName = "****"
  val password = "****"

  def constructElasticsearchParams(indexName: String, typeName: String, apiVersion: ApiVersion) =
    if (apiVersion eq ApiVersion.V5)
      ElasticsearchParams.V5(indexName, typeName)
    else if (apiVersion eq ApiVersion.V7)
      ElasticsearchParams.V7(indexName)
    else
      throw new IllegalArgumentException("API version " + apiVersion + " is not supported")

  val connectionSettings = ElasticsearchConnectionSettings
    .create(connectionString).withCredentials(userName, password)

  val sinkSettings =
   ElasticsearchWriteSettings.create(connectionSettings).withApiVersion(ApiVersion.V7);

  implicit val formatVersionTestDoc: JsonFormat[VolResult] = jsonFormat3(VolResult)

  Source(List(VolResult("test" , 1 , System.currentTimeMillis())))
    .map { message: VolResult =>
      WriteMessage.createIndexMessage("00002", message )
    }
    .log(("Error"))
   .runWith(
     ElasticsearchSink.create[VolResult](
        constructElasticsearchParams("ccy_vol_normalized", "_doc", ApiVersion.V7),
        settings = sinkSettings
      )
    )

}

输出：

19:15:51.815 [default-akka.actor.default-dispatcher-5] INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started
19:15:52.547 [default-akka.actor.default-dispatcher-5] ERROR akka.stream.alpakka.elasticsearch.impl.ElasticsearchSimpleFlowStage$StageLogic - Received error from elastic after having already processed 0 documents. Error: java.lang.RuntimeException: Request failed for POST /_bulk

我是否正确定义了case类DataPayload？它是否与索引Map中定义的预期负载匹配？：

"properties": {
"timestamp": { "type": "date",
"format": "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
},
"vol": { "type": "float" },
"symbol": { "type": "text" }
}

使用Elastic dev工具，以下命令将成功插入文档：

POST ccy_vol_normalized/_doc/
{
"timestamp": "2022-10-21T00:00:00.000Z",
"vol": 1.221,
"symbol" : "SYM"
}

elasticsearch

来源：https://stackoverflow.com/questions/74173097/inserting-a-document-into-elastic-using-alpakka

1条答案

按热度按时间

wixjitnu1#

这是可行的：

import akka.actor.ActorSystem
import akka.stream.alpakka.elasticsearch._
import akka.stream.alpakka.elasticsearch.scaladsl.ElasticsearchSink
import akka.stream.scaladsl.Source
import spray.json.DefaultJsonProtocol._
import spray.json.JsonFormat

import java.text.SimpleDateFormat
import java.util.Date

object AlpakkaWrite extends App {
  val connectionString = "";

  implicit val actorSystem = ActorSystem()
  val userName = ""
  val password = ""
  val connectionSettings = ElasticsearchConnectionSettings
    .create(connectionString).withCredentials(userName, password)
  val sinkSettings =
    ElasticsearchWriteSettings.create(connectionSettings).withApiVersion(ApiVersion.V7);

  val HOUR = 1000 * 60 * 60
  val utcDate = new Date(System.currentTimeMillis() - HOUR)
  val ts = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS").format(utcDate) + "Z"

  implicit val formatVersionTestDoc: JsonFormat[VolResult] = jsonFormat3(VolResult)

  def constructElasticsearchParams(indexName: String, typeName: String, apiVersion: ApiVersion) =
    if (apiVersion eq ApiVersion.V5)
      ElasticsearchParams.V5(indexName, typeName)
    else if (apiVersion eq ApiVersion.V7)
      ElasticsearchParams.V7(indexName)
    else
      throw new IllegalArgumentException("API version " + apiVersion + " is not supported")

  case class VolResult(symbol: String, vol: Double, timestamp: String)
  println("ts : " + ts)

  Source(List(VolResult("test1", 1, ts)))
    .map { message: VolResult =>
      WriteMessage.createIndexMessage(System.currentTimeMillis().toString, message)
    }
    .log(("Error"))
    .runWith(
      ElasticsearchSink.create[VolResult](
        constructElasticsearchParams("ccy_vol_normalized", "_doc", ApiVersion.V7),
        settings = sinkSettings
      )
    )

}

我的日期格式不正确，使用：

val HOUR = 1000 * 60 * 60
  val utcDate = new Date(System.currentTimeMillis() - HOUR)
  val ts = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS").format(utcDate) + "Z"

已修复此问题。

赞(0）回复(0）举报 2022-11-02

我来回答

elasticsearch 使用Alpakka将文档插入Elastic

1条答案

相关问题

热门标签

最新问答