Разъем Spark Cassandra с Java для чтения - PullRequest
Новая
       20

Разъем Spark Cassandra с Java для чтения

0 голосов
/ 10 июля 2020

Требование: - Я сохранил данные в cassandra, и мне нужно ежечасно рассчитывать некоторую оценку на основе обновлений, происходящих с записями. Я вижу, что данные поступают правильно, используя метод show () для набора данных

Ниже код для чтения данных: - 

Dataset<DealFeedSchema> dealFeedSchemaDataset = session.read()
     .format(Constants.SPARK_CASSANDRA_SOURCE_PATH)
     .option(Constants.KEY_SPACE, Constants.CASSANDRA_KEY_SPACE)
     .option(Constants.TABLE, Constants.CASSANDRA_DEAL_TABLE_SPACE)
     .option(Constants.DATE_FORMAT, "yyyy-MM-dd HH:mm:ss")
     .schema(DealFeedSchema.getDealFeedSchema())
     .load()
     .as(Encoders.bean(DealFeedSchema.class));
dealFeedSchemaDataset.show();

вывод шоу ниже: 

+-------+----------+-------------+--------------------+-----------+------------+----------+------------------------+---------------+-----------+-------------------+-------------------+----------+------------+-------------------+----------------+-------------------+-------------+----------+--------------------+----------+-------------------------+---------------+----------------+---------------+--------------+--------------+-----+
|deal_id| deal_name|deal_category|           deal_tags|growth_tags|deal_tag_ids|deal_price|deal_discount_percentage|deal_group_size|deal_active|    deal_start_time|        deal_expiry|product_id|product_name|product_description|product_category|product_category_id|product_price|hero_image|      product_images| video_url|video_thumbnail_image_url|deal_like_count|deal_share_count|deal_view_count|deal_buy_count|weighted_score|boost|
+-------+----------+-------------+--------------------+-----------+------------+----------+------------------------+---------------+-----------+-------------------+-------------------+----------+------------+-------------------+----------------+-------------------+-------------+----------+--------------------+----------+-------------------------+---------------+----------------+---------------+--------------+--------------+-----+
|      4|7h12349961|          mqw|[under999, under3...|         []|          []|    4969.0|                    null|       95166551|          1|2020-07-08 14:48:57|2020-07-18 14:48:57|4725457233|  kao62ggnm7|         32h64e356z|      jnnh29zr1f|               null|       6651.0|86kk7s34yr|[dSt4P79, i4WXOHb...|d6tag27924|               4j1l36lp17|           null|            null|           null|          null|          null| null|

Итак, вот кое-что, что происходит, когда я использую map / foreach для dealFeedSchemaDataset данные кажутся неправильными, я получаю значение столбца deal_start_time в качестве текущего системного времени, как показано ниже, не знаю, как это изменится.

даже строка ниже дает ту же проблему: 

dealFeedSchemaDataset.select(
      functions.col("deal_start_time")).as(Encoders.bean(DateTime.class))
.collectAsList().forEach(schema -> System.out.println(schema));

2020-07-10T20:21:47.895+05:30

может кто-нибудь помочь мне с тем, что я делаю не так?

1 Ответ

0 голосов
/ 19 июля 2020 
java.sql.Timestamp

Это для использования форматов, которые содержат часть времени 

java.sql.Date

Это только для дат

...