将UnixTimestamp转换为Cassandra的TIMEUUID

Question

我正在学习关于Apache Cassandra 3.x.x的所有知识，并试图开发一些可以玩的东西。问题是，我希望将数据存储到包含以下列的Cassandra表中：

id (UUID - Primary Key) | Message (TEXT) | REQ_Timestamp (TIMEUUID) | Now_Timestamp (TIMEUUID)

创建Now_Timestamp很容易，我只需使用now（）函数，它就会自动生成TIMEUUID。问题出现在req_timestamp。如何将Unix时间戳转换为TIMEUUID，以便Cassandra可以存储它？这可能吗？

我的后端架构是这样的：我将JSON中的数据从前端获取到处理它的web服务，并将其存储在Kafka中。然后，一个火花流作业将Kafka日志放入卡桑德拉。

这是我的WebService将数据放入Kafka中。

@Path("/")
public class MemoIn {

    @POST
    @Path("/in")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.TEXT_PLAIN)
    public Response goInKafka(InputStream incomingData){
        StringBuilder bld = new StringBuilder();
        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(incomingData));
            String line = null;
            while ((line = in.readLine()) != null) {
                bld.append(line);
            }
        } catch (Exception e) {
            System.out.println("Error Parsing: - ");
        }
        System.out.println("Data Received: " + bld.toString());

        JSONObject obj = new JSONObject(bld.toString());
        String line = obj.getString("id_memo") + "|" + obj.getString("id_writer") +
                                 "|" + obj.getString("id_diseased")
                                 + "|" + obj.getString("memo") + "|" + obj.getLong("req_timestamp");

        try {
            KafkaLogWriter.addToLog(line);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return Response.status(200).entity(line).build();
    }


}

    package main.java.vcemetery.webservice;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;

public class KafkaLogWriter {

    public static void addToLog(String memo)throws Exception {
        // private static Scanner in;
            String topicName = "MemosLog";

            /*
            First, we set the properties of the Kafka Log
             */
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

            // We create the producer
            Producer<String, String> producer = new KafkaProducer<>(props);
            // We send the line into the producer
            producer.send(new ProducerRecord<>(topicName, memo));
            // We close the producer
            producer.close();

    }
}

public class MemoStream {

    public static void main(String[] args) throws Exception {
        Logger.getLogger("org").setLevel(Level.ERROR);
        Logger.getLogger("akka").setLevel(Level.ERROR);

        // Create the context with a 1 second batch size
        SparkConf sparkConf = new SparkConf().setAppName("KafkaSparkExample").setMaster("local[2]");
        JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(10));

        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "localhost:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "group1");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        /* Se crea un array con los tópicos a consultar, en este caso solamente un tópico */
        Collection<String> topics = Arrays.asList("MemosLog");

        final JavaInputDStream<ConsumerRecord<String, String>> kafkaStream =
                KafkaUtils.createDirectStream(
                        ssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );

        kafkaStream.mapToPair(record -> new Tuple2<>(record.key(), record.value()));
        // Split each bucket of kafka data into memos a splitable stream
        JavaDStream<String> stream = kafkaStream.map(record -> (record.value().toString()));
        // Then, we split each stream into lines or memos
        JavaDStream<String> memos = stream.flatMap(x -> Arrays.asList(x.split("\n")).iterator());
        /*
         To split each memo into sections of ids and messages, we have to use the code \\ plus the character
          */
        JavaDStream<String> sections = memos.flatMap(y -> Arrays.asList(y.split("\\|")).iterator());
        sections.print();
        sections.foreachRDD(rdd -> {
           rdd.foreachPartition(partitionOfRecords -> {
               //We establish the connection with Cassandra
               Cluster cluster = null;
               try {
                   cluster = Cluster.builder()
                           .withClusterName("VCemeteryMemos") // ClusterName
                           .addContactPoint("127.0.0.1") // Host IP
                           .build();

               } finally {
                   if (cluster != null) cluster.close();
               }
               while(partitionOfRecords.hasNext()){


               }
           });
        });

        ssc.start();
        ssc.awaitTermination();

    }
}

提前谢谢你。

Answer 1

Cassandra没有从UNIX时间戳转换的函数。您必须在客户端进行转换。

参考:https://docs.datastax.com/en/cql/3.3/cql/cql_reference/timeuuid_functions_r.html