API en temps réel dans le contexte d'Apache Kafka

Bonjour à tous. Déjà en décembre, OTUS lancera un nouveau volet du cours Architecte logiciel . A la veille du début du cours, je souhaite partager avec vous la traduction d'un article intéressant. Je suggère également de regarder un enregistrement de cours de démonstration sur le sujet: "Idempotence et commutabilité de l'API dans les files d'attente et HTTP" .

---

L'une des questions délicates auxquelles nous sommes constamment confrontés lors de la conception d'applications et de systèmes en général est de savoir comment organiser efficacement l'échange d'informations entre les composants, tout en conservant la flexibilité de changer d'interface sans affecter indûment les autres parties du système. Plus une interface est spécifique et optimisée, plus elle est susceptible d'être si situationnelle qu'elle nécessitera une réécriture complète pour la modifier. Et vice versa; Les génériques peuvent être assez réactifs et largement pris en charge, hélas, au détriment des performances.

(Events) , API (real-time APIs) , , ; , .

. , , - , , . : , , .

(state) - , NoSQL . , , - - . , (consumers) - , . (producers) , , .

, , , , . , . (payload) , , . :

• : userLogin
  
  
  
  

  
  
  
  
  

  • : zbeeblebrox



2020-08-17 16:26:39 BST
    
    
    
    

    
    
    
    
    

• : CarParked
  
  
  
  

  
  
  
  
  

  • : A42 XYZ



2020-08-17 16:36:27



X42
    
    
    
    

    
    
    
    
    

• : orderPlaced
  
  
  
  

  
  
  
  
  

  • :
    
    
    
    
    
    
    
    £2.25



2020-08-17 16:35:41 BST
    
    
    
    

    
    
    
    
    
    
    
    
    
    

(, , ), (, , , , ).

, , , , - Apache Kafka®. Kafka - , :

• Pub/Sub

  
  
  
  
  

  • () () , / .

    
    
    
    
    

• 
  
  
  
  

  • , .

    
    
    
    
    

• 
  
  
  
  

  • .

    
    
    
    
    

Kafka . , , , , , . , , , , NoSQL .

API-, Apache Kafka, , , .

API

, Kafka, , , , , , ( - ). , Kafka - . , , - Kafka (topic - Kafka, - ), .

Kafka . Kafka Go:

package main

import (
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func main() {

    topic := "test_topic"
    p, _ := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092"})
    defer p.Close()

    p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic,
            Partition: 0},
        Value: []byte("Hello world")}, nil)

}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka , , , , , ( ).

, Kafka, . pub/sub, , , . Kafka , , , A/B-, , . , , . , RabbitMQ, ActiveMQ, .

package main

import (
    "fmt"

    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func main() {

    topic := "test_topic"

    cm := kafka.ConfigMap{
        "bootstrap.servers":        "localhost:9092",
        "go.events.channel.enable": true,
        "group.id":                 "rmoff_01"}

    c, _ := kafka.NewConsumer(&cm)
    defer c.Close()
    c.Subscribe(topic, nil)

    for {
        select {
        case ev := <-c.Events():
            switch ev.(type) {

            case *kafka.Message:
                km := ev.(*kafka.Message)
                fmt.Printf("✅ Message '%v' received from topic '%v'\n", string(km.Value), string(*km.TopicPartition.Topic))
            }
        }
    }

}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka, (Consumer Group). . -, Kafka , , , . -, - , , . Kafka , (, ).

, - . Kafka Connect API, .

Producer Consumer API Java, C/C++, Go, Python, Node.js . , HTTP Kafka? REST Proxy.

REST API Apache Kafka

, . , , , :

{
    "name": "NCP Sheffield",
    "space": "A42",
    "occupied": true
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka, . Kafka Confluent REST Proxy - REST-:

curl -X POST \
     -H "Content-Type: application/vnd.kafka.json.v2+json" \
     -H "Accept: application/vnd.kafka.v2+json" \
     --data '{"records":[{"value":{ "name": "NCP Sheffield", "space": "A42", "occupied": true }}]}' \
     "http://localhost:8082/topics/carpark"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, Consumer API, , REST-. Consumer API, , REST API, Consumer Group, (subscription). , REST API , :

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \
      --data '{"name": "rmoff_consumer", "format": "json", "auto.offset.reset": "earliest"}' \
      http://localhost:8082/consumers/rmoff_consumer

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["carpark"]}' \
 http://localhost:8082/consumers/rmoff_consumer/instances/rmoff_consumer/subscription
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \
      http://localhost:8082/consumers/rmoff_consumer/instances/rmoff_consumer/records
[
    {
        "topic": "carpark",
        "key": null,
        "value": {
            "name": "Sheffield NCP",
            "space": "A42",
            "occupied": true
        },
        "partition": 0,
        "offset": 0
    }
]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . , REST-.

, Kafka. , pub/sub. - , , ? , , .

,

Apache Kafka pub/sub - iPhone . , … . Apache Kafka Kafka Streams API. Java Kafka . Kafka Streams, , Walmart, Ticketmaster Bloomberg, ksqlDB.

ksqlDB - , . API SQL Kafka. ksqlDB , , - .

ksqlDB :

CREATE STREAM CARPARK_EVENTS (NAME     VARCHAR,
                              SPACE    VARCHAR,
                              OCCUPIED BOOLEAN)
                        WITH (KAFKA_TOPIC='carpark',
                              VALUE_FORMAT='JSON');
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB , , . , . , , , :

SELECT TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM-dd HH:mm:ss') AS EVENT_TS,
       SPACE
  FROM CARPARK_EVENTS
 WHERE NAME='Sheffield NCP'
   AND OCCUPIED=false
  EMIT CHANGES;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

SQL-, , , , ( EMIT CHANGES). , push-, , , . ksqlDB pull- ( ), , , . , ksqlDB , , .

ksqlDB REST API, SQL :

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT TIMESTAMPTOSTRING(ROWTIME,'\''yyyy-MM-dd HH:mm:ss'\'') AS EVENT_TS, SPACE FROM CARPARK_EVENTS WHERE NAME='\''Sheffield NCP'\'' and OCCUPIED=false EMIT CHANGES;"}'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, :

{"queryId":"383894a7-05ee-4ec8-bb3b-c5ad39811539","columnNames":["EVENT_TS","SPACE"],"columnTypes":["STRING","STRING"]}
…
["2020-08-05 16:02:33","A42"]
…
…
…
["2020-08-05 16:07:31","D72"]
…
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB . SELECT



CREATE STREAM streamname AS



Kafka. , ksqlDB , , .. , Kafka. ksqlDB , , :

CREATE STREAM CARPARKS AS
    SELECT E.NAME AS NAME, E.SPACE,
           R.LOCATION, R.CAPACITY,
           E.OCCUPIED,
           CASE
               WHEN OCCUPIED=TRUE THEN 1
               ELSE -1
           END AS OCCUPIED_IND
    FROM   CARPARK_EVENTS E
           INNER JOIN
           CARPARK_REFERENCE R
           ON E.NAME = R.NAME;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, CASE , . CREATE STREAM Kafka, :

+----------------+-------+----------+----------------------------+----------+--------------+
|NAME            |SPACE  |OCCUPIED  |LOCATION                    |CAPACITY  |OCCUPIED_IND  |
+----------------+-------+----------+----------------------------+----------+--------------+
|Sheffield NCP   |E48    |true      |{LAT=53.4265964, LON=-1.8426|1000      |1             |
|                |       |          |386}                        |          |              |
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , ksqlDB . , SQL, :

CREATE TABLE CARPARK_SPACES AS
    SELECT NAME,
           SUM(OCCUPIED_IND) AS OCCUPIED_SPACES
        FROM CARPARKS
        GROUP BY NAME;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB REST API:

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT OCCUPIED_SPACES FROM CARPARK_SPACES WHERE NAME='\''Birmingham NCP'\'';"}'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , ( "pull- ", " push- ") , :

{"queryId":null,"columnNames":["OCCUPIED_SPACES"],"columnTypes":["INTEGER"]}
[30]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, -

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT OCCUPIED_SPACES FROM CARPARK_SPACES WHERE NAME='\''Birmingham NCP'\'';"}'
{"queryId":null,"columnNames":["OCCUPIED_SPACES"],"columnTypes":["INTEGER"]}
[29]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Java ksqlDB Python Go.

Apache Kafka , , , ( ), .

, , , , , :

• 
  
  
  
  

• 
  
  
  
  

• 
  
  
  
  

Apache Kafka Connect API, , Kafka. , Kafka S3 :

curl -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/sink-s3/config \
    -d ' {
        "connector.class": "io.confluent.connect.s3.S3SinkConnector",
        "topics": "carpark",
        "s3.bucket.name": "rmoff-carparks",
        "s3.region": "us-west-2",
        "flush.size": "1024",
        "storage.class": "io.confluent.connect.s3.storage.S3Storage",
        "format.class": "io.confluent.connect.s3.format.json.JsonFormat"
        }'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , , S3. . , Snowflake, Kafka Connect; . Kafka Connect Kafka. , CARPARK_REFERENCE



, ksqlDB , (CDC - change data capture) , .

Apache Kafka , . , , , , , .

API Kafka , ksqlDB, , / . API , REST.

Apache Kafka, developer.confluent.io. Confluent Platform - Apache Kafka, , . , Confluent Cloud.  GitHub Docker Compose , . - , Kafka, « - ».

---

- : " API HTTP".

---