API Real Time dalam konteks Apache Kafka

Halo semuanya. Sudah pada bulan Desember, OTUS akan meluncurkan aliran baru dari kursus Arsitek Perangkat Lunak . Untuk mengantisipasi dimulainya kursus, saya ingin berbagi dengan Anda terjemahan dari artikel yang menarik. Saya juga menyarankan untuk menonton rekaman demo-pelajaran tentang topik: "Idempotensi dan pergantian API dalam antrian dan HTTP" .

---

Salah satu pertanyaan rumit yang selalu kami hadapi saat merancang aplikasi dan sistem secara umum adalah bagaimana mengatur pertukaran informasi antar komponen secara efektif, sambil mempertahankan fleksibilitas untuk mengubah antarmuka tanpa terlalu memengaruhi bagian lain dari sistem. Semakin spesifik dan optimal sebuah antarmuka, semakin besar kemungkinannya untuk menjadi begitu situasional sehingga mengubahnya membutuhkan penulisan ulang yang lengkap. Dan sebaliknya; Generik bisa sangat adaptif dan didukung secara luas, tetapi sayangnya, dengan mengorbankan kinerja.

(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".

---