我们配置了jdbc logstash，通过logstash将数据从cassandra迁移到elasticsearch。我们已经能够通过logstash将数据从cassandra移动到elasticsearch，我们还在基于文档id更新elasticsearch中的数据
我们希望根据cassandra的更改进行增量更新，所以我们在logstash input jdbc中使用sql\u last\u值

statement => "SELECT emp_id,emp_city,emp_name,emp_phone,emp_sal,updated_time FROM cloud.employee_list where updated_time > :sql_last_value and updated_time < toTimestamp(now()) ALLOW FILTERING;"

[ERROR] 2020-11-05 19:00:00.517 [Ruby-0-Thread-26: :1] jdbc - Java::JavaSql::SQLException: [Simba][CassandraJDBCDriver](500211) ERROR Invalid query: SELECT emp_id,emp_city,emp_name,emp_phone,emp_sal,updated_time FROM iqcloud.employee_list where updated_time > '2020-09-09 07:57:12.836000' and updated_time < toTimestamp(now()) ALLOW FILTERING, Cause: com.simba.cassandra.shaded.datastax.driver.core.exceptions.InvalidQueryException: Unable to coerce '2020-09-09 07:57:12.836000' to a formatted date (long).: SELECT emp_id,emp_city,emp_name,emp_phone,emp_sal,updated_time FROM cloud.employee_list where updated_time > '2020-09-09 07:57:12.836000' and updated_time < toTimestamp(now()) ALLOW FILTERING;

在cassandra timestamp列中，存储的时间大约在2020-09-09 07:57:12.836000+0000以下，当我们需要获取时，我们应该删除额外的零并获取如下更新的时间='2020-09-09 07:57:12.836'
在错误日志中，由于sql\u last\u值中有额外的零，我们得到的查询无效。有没有任何方法可以在sql\u last\u value中使用我们的timestamp列值，或者其他任何方法来实现sql\u last\u value？
有人能澄清一下吗？