حالت تاریک
فصل ۱۱ — Schema Registry
این فصل شامل
- توسعه مدل بلوغ پیشنهادی Kafka
- ارزشی که schemaها هنگام تغییر داده میدهند
- مرور Avro و serialization داده
- قوانین سازگاری برای تغییر schema در طول زمان
با کشف راههای مختلف استفاده از Apache Kafka، جالب است فکر کنید با استفاده بیشتر نگاهتان به Kafka چگونه عوض میشود. با رشد سازمانها (یا حتی ابزارها)، گاهی میتوان آنها را با سطوح بلوغ مدل کرد. Martin Fowler توضیح خوبی در https://martinfowler.com/bliki/MaturityModel.html [۱] دارد. Fowler مثال Richardson Maturity Model برای REST [۲] را هم دارد. برای مرجع بیشتر، سخنرانی اصلی Leonard Richardson «Justice Will Take Us Millions Of Intricate Moves: Act Three: The Maturity Heuristic» در https://www.crummy.com/writing/speaking/2008-QCon/act3.html¹
¹ متن وبسایت act3.html تحت Creative Commons License در https://creativecommons.org/licenses/by-sa/2.0/legalcode است.
۱۱.۱ مدل بلوغ پیشنهادی Kafka
در بخشهای بعد روی سطوح بلوغ مخصوص Kafka تمرکز میکنیم. برای مقایسه، white paper Confluent «Five Stages to Streaming Platform Adoption» دیدگاه دیگری با پنج مرحله مدل بلوغ streaming و معیارهای متمایز هر مرحله [۳] ارائه میدهد. اولین سطح را میبینیم (البته بهعنوان programmer از سطح ۰ شروع میکنیم).
این تمرین با مدل بلوغ کمک میکند فکر کنیم Kafka چطور ابزار قدرتمند یک اپلیکیشن یا حتی پایه همه اپلیکیشنهای سازمان — نه فقط message broker ساده — میشود. این سطوح مسیر گامبهگام اجباری نیستند، بلکه راهی برای فکر کردن به شروع و پیشرفت با Kafka هستند. البته قابل بحثاند؛ فقط یک مسیر نمونه پیشنهاد میکنیم.
۱۱.۱.۱ سطح ۰
در این سطح Kafka را بهعنوان enterprise service bus (ESB) یا سیستم publish/subscribe (pub/sub) استفاده میکنیم. رویدادها ارتباط ناهمگام بین اپلیکیشنها را فراهم میکنند، چه message broker دیگری مثل RabbitMQ را جایگزین کنیم چه از همین الگو شروع کنیم.
یک use case: کاربر سند متنی برای تبدیل به PDF ارسال میکند. پس از ارسال، اپلیکیشن سند را ذخیره و پیامی به topic Kafka میفرستد. consumer Kafka پیام را میخواند تا مشخص کند کدام اسناد باید به PDF تبدیل شوند. در این مثال، بار به backend منتقل میشود که کاربر میداند فوراً پاسخ نمیدهد. شکل ۱۱.۱ message bus را نشان میدهد.
PDF generation
at a later time
Request
response Consumer app
Web UI Kafka
to PDF
PDF
PDF
PDFشکل ۱۱.۱ مثال سطح ۰
همین سطح مزیت decouple کردن سیستم را میدهد: خرابی frontend ارسال متن backend را تحت تأثیر نمیگذارد و لازم نیست هر دو همزمان موفق بمانند.
۱۱.۱.۲ سطح ۱
batch processing هنوز در بخشهایی از سازمان ممکن است باشد، اما بیشتر داده تولیدشده به Kafka میآید. با ETL یا change data capture (CDC)، Kafka رویداد را از سیستمهای بیشتری جمع میکند. سطح ۱ جریان داده operational و real-time و تغذی سریع سیستمهای تحلیلی را میدهد.
مثال: پایگاه vendor با اطلاعات مشتری. نمیخواهیم marketing کوئریهای سنگین بزند و ترافیک production را کند کند. با Kafka Connect داده جداول را به topicهای Kafka مینویسیم و روی شرایط خودمان استفاده میکنیم. شکل ۱۱.۲ Kafka Connect را در capture از relational database نشان میدهد.
Vendor database Kafka connect
Kafka
Other Marketing
consumers consumer
Serving customer Pulls database
operational traffic changes into
Kafkaشکل ۱۱.۲ مثال سطح ۱
۱۱.۱.۳ سطح ۲
میفهمیم داده با زمان تغییر میکند و schema لازم است. producer و consumer decouple هستند، اما هنوز باید داده را بفهمند. از schema و schema registry استفاده میکنیم. ایدهآل بود از schema شروع کنیم، اما واقعیت این است که این نیاز اغلب چند تغییر اپلیکیشن بعد از deploy اولیه خود را نشان میدهد.
مثال: تغییر ساختار رویداد سفارش از سیستم پردازش. فیلد جدید اضافه میشود اما optional است و چون schema registry برای backward compatibility پیکربندی شده، کار میکند. شکل ۱۱.۳ نیاز consumer به schema را نشان میدهد.
Order 1 Same Kafka topic
Field 0
Field 1
Order Consumer
Order 1 Order 2
Message Message
Order 2
Field 0
Field 1
Field 2 Consumers need to know how to
Added, but optional handle both message versions.شکل ۱۱.۳ مثال سطح ۲
۱۱.۱.۴ سطح ۳
همه چیز event stream بیپایان است. Kafka سیستم سازمان برای اپلیکیشنهای event-based است. مشتری منتظر recommendation یا گزارش وضعیتی که قبلاً batch شبانه تولید میشد نمیماند؛ در میلیثانیه از تغییر حساب با وقوع رویداد مطلع میشود. بهجای pull از منابع دیگر، اپلیکیشنها مستقیم از کلاستر pull میکنند. اپلیکیشنهای user-facing بسته به نیاز زیرساخت Kafka اصلی state و materialized view مشتق میکنند.
۱۱.۲ Schema Registry
در این فصل روی سطح ۲ تمرکز میکنیم: برنامهریزی برای تغییر داده. حالا که ارسال و دریافت داده را بلد شدیم — با اشاره کوتاه schema در فصل ۳ — جزئیات مهم جا مانده. Confluent Schema Registry چه میدهد؟
Confluent Schema Registry schemaهای نامگذاریشده را ذخیره و چند نسخه نگه میدارد [۴]. از نظر هدف شبیه Docker Registry است که imageهای Docker را ذخیره و توزیع میکند. چرا storage لازم است؟ producer و consumer به هم وابسته نیستند، اما باید schema داده را از همه کلاینتها کشف کنند. با registry میزبانی remote، کاربران نیازی به کپی محلی یا ساخت registry خود از لیست schema ندارند.
Schemaها نوعی interface برای اپلیکیشنها هستند و از breaking change جلوگیری میکنند [۴]. چرا برای داده سریع مهم است؟ storage و retention Kafka اجازه میدهد consumer به پیامهای قدیمیتر (ماهها یا بیشتر) برگردد و باید نسخههای مختلف داده را handle کند.
برای Kafka از Confluent Schema Registry استفاده میکنیم. Confluent گزینه عالی برای schema است. اگر Kafka را از Confluent Platform قبل از این فصل نصب کردهاید، ابزارها برای کاوش دارید. اگر نه، نصب و راهاندازی registry را در بخشهای بعد میبینیم.
۱۱.۲.۱ نصب Confluent Schema Registry
Confluent Schema Registry نرمافزار community بخشی از Confluent Platform است [۵]. Schema Registry خارج از Kafka Broker زندگی میکند، اما خود از Kafka با topic _schemas [۶] بهعنوان لایه storage استفاده میکند. حذف تصادفی این topic ممنوع!
Schema Registry nodes
REST API Primary Clients
Writes Reads
Kafka
Schemas stored in Kafka
like kinaction_alertشکل ۱۱.۴ زیرساخت Schema Registry
برای production، Schema Registry باید روی سرور جدا از brokerها host شود (شکل ۱۱.۴) [۶]. با سیستم توزیعشده و انتظار failure، چند instance registry ممکن است. همه nodeها lookup از کلاینت و write را به primary route میکنند؛ کلاینتها لیست node خاص نگه نمیدارند.
۱۱.۲.۲ پیکربندی Registry
مثل سایر اجزای Kafka، پارامترهای پیکربندی در فایل تنظیم میشوند. پس از نصب Kafka، پیشفرض در etc/schema-registry/schema-registry.properties است. registry باید بداند schemaها در کدام topic ذخیره شوند و با کلاستر Kafka شما چگونه کار کند.
در listing 11.1 از ZooKeeper برای election primary استفاده میکنیم. فقط primary به topic Kafka مینویsd. اگر تیم از وابستگی ZooKeeper دور میشود، election مبتنی بر Kafka با kafkastore.bootstrap.servers [۷] هم ممکن است.
Listing 11.1 پیکربندی Schema Registry
listeners=http://localhost:8081
kafkastore.connection.url=localhost:2181
kafkastore.topic=_schemas
debug=trueSchema Registry را start میکنیم. ZooKeeper و brokerهای Kafka باید از قبل بالا باشند. سپس اسکریپت start از خط فرمان [۸]:
Listing 11.2 راهاندازی Schema Registry
bash
bin/schema-registry-start.sh \
./etc/schema-registry/schema-registry.propertiesبا jps یا بررسی process (اپلیکیشن Java مثل broker و ZooKeeper) تأیید میکنیم. حالا نحوه استفاده از اجزا — با schema فصل ۳.
۱۱.۳ ویژگیهای Schema
Confluent Schema Registry اجزای مهم زیر را دارد: REST API (و اپلیکیشن زیرین) برای ذخیره و fetch schema؛ کتابخانه کلاینت برای retrieve و مدیریت schema محلی. ابتدا REST API.
۱۱.۳.۱ REST API
REST API این منابع را مدیریت میکند: schema، subject، compatibility و config [۹]. «subject» شاید نیاز به توضیح داشته باشد. نسخهها و خود subjectها را create، retrieve و delete میکنیم. topic و subject مرتبط برای اپلیکیشن با topic kinaction_schematest:
در registry subject kinaction_schematest-value داریم چون نام پیشفرض بر اساس topic فعلی است. اگر schema برای message key هم باشد، kinaction_schematest-key هم داریم. key و value subject جدا [۱۰] هستند تا schema key و value مستقل version و تغییر کنند.
برای تأیید start و دیدن API، GET با curl [۹]. listing 11.3 سطح compatibility را list میکند.
Listing 11.3 گرفتن پیکربندی Schema Registry
bash
curl -X GET http://localhost:8081/configبرای REST با Schema Registry header Content-Type لازم است. در مثالهای بعد مثل listing 11.7 از application/vnd.schemaregistry.v1+json [۹] استفاده میکنیم. مانند schemaها، نسخه API را اعلام میکنیم تا کلاینت نسخه intended را استفاده کند.
REST API برای admin subject و schema عالی است؛ بیشتر developerها با کتابخانه کلاینت وقت میگذارند.
۱۱.۳.۲ کتابخانه کلاینت
تعامل producer با Schema Registry: به مثال فصل ۳ با producer پیکربندیشده برای Avro serializer برگردید. registry محلی باید بالا باشد؛ producer را پیکربندی میکنیم (listing 11.4). در فصل ۳ schema برای شی Alert بهعنوان value پیام ساختیم. value.serializer باید KafkaAvroSerializer باشد. این کلاس شی سفارشی را با Registry serialize میکند.
Listing 11.4 Producer با Avro serialization
java
...
kaProperties.put("key.serializer",
"org.apache.kafka.common.serialization.LongSerializer");
kaProperties.put("value.serializer",
"io.confluent.kafka.serializers.KafkaAvroSerializer");
kaProperties.put("schema.registry.url",
"http://localhost:8081");
Producer<Long, Alert> producer =
new KafkaProducer<Long, Alert>(kaProperties);
Alert alert = new Alert();
alert.setSensorId(12345L);
alert.setTime(Calendar.getInstance().getTimeInMillis());
alert.setStatus(alert_status.Critical);
log.info("kinaction_info = {}, alert.toString()");
ProducerRecord<Long, Alert> producerRecord =
new ProducerRecord<Long, Alert>(
"kinaction_schematest", alert.getSensorId(), alert
);
producer.send(producerRecord);NOTE با TopicNameStrategy پیشفرض، Schema Registry subject
kinaction_schematest-valueرا با schema Alert ثبت میکند. برای strategy دیگر:value.subject.name.strategyوkey.subject.name.strategy[۱۰].
در consumer، پس از یافتن schema، recordها فهمیده میشوند. listing 11.5 همان schema را میخواند [۱۱].
Listing 11.5 Consumer با Avro deserialization
java
kaProperties.put("key.deserializer",
"org.apache.kafka.common.serialization.LongDeserializer");
kaProperties.put("value.deserializer",
"io.confluent.kafka.serializers.KafkaAvroDeserializer");
kaProperties.put("schema.registry.url",
"http://localhost:8081");
...
KafkaConsumer<Long, Alert> consumer =
new KafkaConsumer<Long, Alert>(kaProperties);
consumer.subscribe(List.of("kinaction_schematest"));
while (keepConsuming) {
ConsumerRecords<Long, Alert> records =
consumer.poll(Duration.ofMillis(250));
for (ConsumerRecord<Long, Alert> record : records) {
log.info("kinaction_info Alert Content = {}, record.value().toString()");
}
}تا اینجا فقط یک نسخه schema با producer و consumer کار کردیم. برنامهریزی برای تغییر داده سردرد زیاد جلوگیری میکند. قوانین تغییرات مجاز و تأثیر بر کلاینتها را میبینیم.
۱۱.۴ قوانین سازگاری (Compatibility)
مهم است چه استراتژی compatibility پشتیبانی میکنید. قوانین این بخش schema در حال تغییر را هدایت میکنند. انواع زیاد به نظر میرسد، اما انواع transitive همان قوانین بدون پسوند را دارند. non-transitive فقط با آخرین نسخه schema چک میشود؛ transitive با همه نسخههای قبلی [۱۲]. انواع Confluent: BACKWARD (پیشفرض)، BACKWARD_TRANSITIVE، FORWARD، FORWARD_TRANSITIVE، FULL، FULL_TRANSITIVE و NONE [۱۲].
BACKWARD برای اپلیکیشنها یعنی چه؟ تغییرات backward-compatible: افزودن فیلد non-required یا حذف فیلد [۱۲]. هنگام انتخاب نوع compatibility، ترتیب upgrade کلاینت مهم است. برای BACKWARD معمولاً consumerها اول upgrade میشوند [۱۲] — باید پیام را قبل از variation جدید بخوانند.
FORWARD برعکس backward است: فیلد جدید اضافه میشود و برعکس BACKWARD، producerها اول upgrade میشوند [۱۲].
تغییر schema Alert برای backward compatibility: listing 11.6 فیلد recovery_details با default «Analyst recovery needed» برای پیامهای بدون مقدار.
Listing 11.6 تغییر schema Alert
json
{"name": "Alert",
...
"fields": [
{"name": "sensor_id", "type": "long",
"doc":"The unique id that identifies the sensor"},
...
{"name": "recovery_details", "type": "string",
"default": "Analyst recovery needed"}
]
}پیامهای قدیمی نسخه ۱ مقدار default برای فیلد بعداً اضافهشده میگیرند. consumer با Schema Registry نسخه ۲ میخواند [۱۲].
۱۱.۴.۱ اعتبارسنجی تغییرات schema
اگر تست API endpoint یا Swagger دارید، automate کردن تست تغییر schema را در نظر بگیرید. دو گزینه:
- endpointهای compatibility REST API
- Maven plugin برای اپلیکیشن JVM
listing 11.7 مثال REST برای چک compatibility [۱۳]. قبل از چک، نسخه قدیمی schema باید در registry باشد؛ در صورت failure، source code کتاب را ببینید.
Listing 11.7 چک compatibility با REST API Schema Registry
bash
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data '{ "schema": "{ \"type\": \"record\", \"name\": \"Alert\",
\"fields\": [{ \"name\": \"notafield\", \"type\": \"long\" } ]}" }' \
http://localhost:8081/compatibility/subjects/kinaction_schematest-value/versions/latestjson
{"is_compatible":false}Maven plugin اگر Maven و JVM دارید [۱۴]. بخشی از pom.xml در listing 11.8؛ فایل کامل در source code فصل.
Listing 11.8 چک compatibility با Maven plugin Schema Registry
xml
<plugin>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-maven-plugin</artifactId>
<configuration>
<schemaRegistryUrls>
<param>http://localhost:8081</param>
</schemaRegistryUrls>
<subjects>
<kinaction_schematest-value>
src/main/avro/alert_v2.avsc
</kinaction_schematest-value>
</subjects>
<goals>
<goal>test-compatibility</goal>
</goals>
</configuration>
...
</plugin>schemaهای مسیر فایل را میگیرد و با schemaهای ذخیرهشده در Registry مقایسه میکند. goal: mvn schema-registry:test-compatibility.
۱۱.۵ جایگزین schema registry
همه پروژهها با schema یا تغییر داده شروع نمیکنند. راهحل ساده برای تغییر format: produce روی topic دیگر با breaking change. پس از consume format قدیم، consumer در صورت نیاز update و خواندن از topic دیگر. اگر reprocess نمیکنید خوب کار میکند. شکل ۱۱.۵ switch به topic جدید پس از خواندن همه پیامهای قدیم. u1 یعنی «update 1» و u2 «update 2».
Schema 1: kinaction_alert
Producer u1 Topic u1 Consumer u1
Retired after
consumption of
old messages
Schema 2: Updated kinaction_alert
Producer u2 Topic u2 Consumer u2شکل ۱۱.۵ جریان جایگزین
اگر reprocess across format برنامه دارید، topic جدید برای پیامهای transformشده topic اول و پیامهای جدید بسازید. Kafka Streams (فصل ۱۲) در topic-to-topic transformation کمک میکند.
خلاصه
- Kafka از use case ساده تا سیستم اصلی سازمان قابل استفاده است.
- Schemaها نسخهبندی تغییر داده را کمک میکنند.
- Schema Registry (پیشنهاد Confluent، جدا از Kafka) راه کار با schemaهای Kafka است.
- با تغییر schema، قوانین compatibility backward، forward یا full compatibility را مشخص میکنند.
- بدون schema، topicهای مختلف برای نسخههای مختلف داده استفاده میشود.
منابع
- M. Fowler. «Maturity Model.» (August 26, 2014). https://martinfowler.com/bliki/MaturityModel.html (accessed June 15, 2021).
- M. Fowler. «Richardson Maturity Model.» (March 18, 2010). https://martinfowler.com/articles/richardsonMaturityModel.html (accessed June 15, 2021).
- L. Hedderly. «Five Stages to Streaming Platform Adoption.» Confluent white paper (2018). https://www.confluent.io/resources/5-stages-streaming-platform-adoption/ (accessed January 15, 2020).
- «Schema Registry Overview.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/index.html (accessed July 15, 2020).
- «Confluent Platform Licenses: Community License.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/license.html#community-license (accessed August 21, 2021).
- «Running Schema Registry in Production.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/installation/deployment.html#schema-registry-prod (accessed April 25, 2019).
- «Schema Registry Configuration Options.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/installation/config.html#schemaregistry-config (accessed August 22, 2021).
- «Schema Registry and Confluent Cloud.» Confluent documentation (n.d.). https://docs.confluent.io/cloud/current/cp-component/schema-reg-cloud-config.html (accessed August 22, 2021).
- «Schema Registry API Reference.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/develop/api.html (accessed July 15, 2020).
- «Formats, Serializers, and Deserializers.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html (accessed April 25, 2019).
- «On-Premises Schema Registry Tutorial.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/schema_registry_onprem_tutorial.html (accessed April 25, 2019).
- «Schema Evolution and Compatibility.» Confluent Platform. https://docs.confluent.io/current/schema-registry/avro.html#compatibility-types (accessed June 1, 2020).
- «Schema Registry API Usage Examples.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/develop/using.html (accessed August 22, 2021).
- «Schema Registry Maven Plugin.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/schema-registry/develop/maven-plugin.html (accessed July 16, 2020).