我正在从jdbc源读取数据并将其直接写入弹性搜索索引。当我查询ES中的数据时,我看到我的数据帧中的所有时间戳字段都转换为long
看下面的代码
val appName="ExractToolEngine"
val master = "local[2]"
val conf = new SparkConf().setAppName(appName).setMaster(master)
conf.set("es.write.operation", "index")
conf.set("es.mapping.id", "user_id")
conf.set("index.mapper.dynamic", "true")
conf.set("es.mapping.rich.date", "true")
def main(args: Array[String]): Unit = {
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
val srcData = sqlContext.read.format("jdbc").
options(Map("driver"->"com.jdbc.Driver",
"url" -> "jdbc...",
"dbtable"-> "tbl",
"partitionColumn"-> "user_id",
"lowerBound"-> "1",
"upperBound"-> "1000000",
"numPartitions"-> "50"
)
).load()
srcData.filter("user_id>=1 and user_id<=1000000").saveToEs("test_users/sm_1")
}
当我跑srcData.printSchema()
我明白了:
|-- dwh_insert_ts: timestamp (nullable = true)
|-- dwh_update_ts: timestamp (nullable = true)
当我在http://localhost:9200/test_users/_mapping/sm_1上查询索引映射时
我知道了
"properties": {
"dwh_insert_ts": {
"type": "long"
},
"dwh_update_ts": {
"type": "long"
},
是否需要强制弹性搜索来保持时间戳并进行转换?
您可以查看以下ES doc page。
在我看来你的配置是错误的和无用的:
conf.set("es.mapping.rich.date", "true")
正确的名称在here中定义:
es.mapping.date.rich
由于它默认为true
你可能不需要它。
即使在设置了“es.mapping.rich.date”之后我仍然面临着这个问题,“真的”我正在使用带有pyspark的elasticsearch 6.6
df.write.format("org.elasticsearch.spark.sql").option("es.nodes.wan.only","true").option("es.port","9200").option("es.net.ssl","false").option("es.nodes", esURL).option("es.mapping.id", "Tuple_ID").option("es.mapping.rich.date","true").mode("Overwrite").save("readings/sensors")