问在not_x_content_exception中向弹性搜索结果发送JSON

EN

{"api":"POST /n2mt/translateLog","n2mt":true,"params":{"deviceId":"D4E3-4239-9188-92E67D246D13","caller":"NID.IOS","source":"ko","target":"ja","text":"넌 친구랑도 금방 친해질 것 같아","dict":true,"dictDisplay":3,"dictHl":"all","honorific":false,"instant":false,"agree":false,"domainType":"","splitSents":false,"sessionId":"9345-41B9-9632-94852823146E","n2mtResut":null},"result":{"json":{"message":{"result":{"srcLangType":"ko","tarLangType":"ja","translatedText":"あなたは友達ともすぐ親しくなると思う。","engineType":"N2MT","pivot":null,"delay":400,"delaySmt":400},"@type":"response","@service":"proxy","@version":"1.0.0"}},"elapsed":0,"textLen":18},"langDetect":"ko","modelVer":"1.2.12"}

{"error":{"root_cause":[{"type":"not_x_content_exception","reason":"not_x_content_exception: Compressor detection can only be called on some xcontent bytes or compressed xcontent bytes"}],"type":"mapper_parsing_exception","reason":"failed to parse","caused_by":{"type":"not_x_content_exception","reason":"not_x_content_exception: Compressor detection can only be called on some xcontent bytes or compressed xcontent bytes"}},"status":400}

  def makeHttpCall(row: String): Unit = {
    val client = HttpClientBuilder.create().setDefaultCredentialsProvider(credentialsProvider).build()
    val post = new HttpPost(s"$elasticsearchHost:$elasticsearchPort/$indexAndDocType")
    post.addHeader(HttpHeaders.CONTENT_TYPE, "application/json;charset=UTF-8")
    val nameValuePairs = new ArrayList[NameValuePair]()
    nameValuePairs.add(new BasicNameValuePair("JSON", row))
    post.setEntity(new UrlEncodedFormEntity(nameValuePairs))
    val response = client.execute(post)
    val entity = response.getEntity()
    println(Seq(response.getStatusLine.getStatusCode(), response.getStatusLine.getReasonPhrase()))
    val responseAsString = EntityUtils.toString(response.getEntity())
    println(responseAsString)
  }

    resultDF.writeStream
      .option("truncate", false)
      .option("encoding", "UTF-8")
      .outputMode(outputMode)
      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
        batchDF.persist()
        val result = batchDF
          .select("json")
          .collect()
          .map(_.getString(0))
          .mkString("\n")
        makeHttpCall(result)
        println(result)
        batchDF.unpersist()
        ()
      }