این مقاله با توجه به درک اینکه Kafka Connect چیست ، چه کاری انجام می دهد و چگونه می توان این کار را در هر محیطی اجرا کرد ، متمرکز خواهد شد. الهام بخش این مقاله تلاش من برای شناسایی نحوه اجرای ارتباط Kafka با استفاده از ابر تلاقی بود زیرا بیشتر مقالات موجود در وب یا از Confluent Cloud یا یک اسکریپت پوسته برای شروع اتصال استفاده می کنند. من می خواستم یک برنامه جاوا ایجاد کنم که می تواند از منبع باز Kafka Connect استفاده کند و می تواند مانند هر شیشه چربی دیگر میزبانی شود.
رمز منبع
اگر دوست دارید خودتان آن را امتحان کنید یا می خواهید کد منبع من را طی کنید ، می توانید من را کلون کنید [GitHub repository](https://github.com/ankit-sood/kafka-connect-standalone) و از طریق پرونده readme عبور کنید. بنابراین ، بیایید شروع کنیم.
پیش نیازهای
ما از جاوا به عنوان زبان اصلی و Maven به عنوان راه حل مدیریت وابستگی استفاده خواهیم کرد. بنابراین ، برای اجرای این کار ، به جاوا 17 و Maven 3.9 نصب شده در سیستم خود نیاز دارید. در صورت نصب جاوا یا Maven ، می توانید با استفاده از لینک های زیر آن را بارگیری و نصب کنید.
من کافکا را به صورت محلی در سیستم خود اجرا می کنم. اگر می خواهید بدانید که چگونه Kafka را به صورت محلی تنظیم کنید ، لطفاً از این استفاده کنید https://dev.to/ankit_sood_66861d56d8e42a/getting-started-with-kafka-connect-a-comprehensive-guide-4k91(https://kafka.apache.org/quickstart) برای تنظیم آن. همچنین ، من استفاده می کنم [Kafdrop](https://github.com/obsidiandynamics/kafdrop) برای مشاهده کارگزاران کافکا ، مباحث ، پیام ها ، گروه های مصرف کننده و ACL.
مقدمه
Kafka Connect ابزاری است که به انتقال داده ها بین Apache Kafka و سایر سیستم ها به روشی قابل اعتماد و مقیاس پذیر کمک می کند. تنظیم اتصالات که مقادیر زیادی از داده ها را در داخل و خارج از کافکا منتقل می کنند ، آسان می شود. می توانید از آن برای وارد کردن داده ها از کل پایگاه داده ها یا جمع آوری معیارها از سرورها به موضوعات Kafka برای پردازش در زمان واقعی استفاده کنید. همچنین اجازه می دهد تا داده های صادرات از Kafka به ذخیره سازی ، سیستم های پرس و جو یا پردازش دسته ای برای تجزیه و تحلیل آفلاین. Kafka Connect انواع مختلفی از اطلاعات را در مورد پیکربندی اتصال IE از اتصال ، وضعیت کار اتصال و آخرین جبران متعهد از اتصال ذخیره می کند.
Kafka Connect در حال حاضر از دو حالت اجرای پشتیبانی می کند:
- مستقل (فرآیند واحد)
- توزیع شده
در حالت مستقل ، تمام کارها در یک فرآیند واحد انجام می شود و جزئیات مربوط به پیکربندی ، جبران و وضعیت در یک پرونده حفظ می شود. این پیکربندی برای تنظیم و شروع کار ساده تر است و ممکن است در شرایطی مفید باشد که فقط یک کارگر منطقی باشد (به عنوان مثال جمع آوری پرونده های ورود به سیستم) ، اما از برخی از ویژگی های اتصال Kafka مانند تحمل گسل سود نمی برد.
در حالت توزیع شده ، Kafka Connect جبران ، پیکربندی ها و وضعیت کار را در موضوعات Kafka ذخیره می کند. توصیه می شود به منظور دستیابی به تعداد پارتیشن ها و عوامل تکثیر ، به صورت دستی مباحث مربوط به جبران ، پیکربندی ها و وضعیت ها را ایجاد کنید.
برای هدف از این تمرین ، ما فقط Kafka Connect خود را اجرا خواهیم کرد حالت توزیع شدهبشر از این رو ، ما مباحث افست ، پیکربندی و وضعیت خود را از قبل ایجاد خواهیم کرد. ما برای ایجاد این موارد از اسکریپت در زیر استفاده خواهیم کرد.
- Connect-Offsets: موضوع برای ذخیره جبران استفاده می شود. این موضوع باید پارتیشن های زیادی داشته باشد و تکرار و فشرده شود.
# offset.storage.topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic connect-offsets --replication-factor 3 --partitions 1 --config cleanup.policy=compact
- Connect-Configs: موضوع برای ذخیره کانکتور و تنظیمات کار استفاده می شود. توجه داشته باشید که این باید یک موضوع واحد ، بسیار تکثیر و فشرده باشد.
# config.storage.topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic connect-configs --replication-factor 1 --partitions 1 --config cleanup.policy=compact
- اتصال به وضعیت: موضوع برای استفاده برای ذخیره وضعیت. این موضوع می تواند چندین پارتیشن داشته باشد و باید تکرار و فشرده شود.
# status.storage.topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic connect-status --replication-factor 1 --partitions 10 --config cleanup.policy=compact
در تصویر فوق می بینیم که مباحث مورد نیاز ایجاد می شود. مرحله بعدی ایجاد یک پروژه جاوا نمونه است. من از Intelij خود برای ایجاد آن استفاده می کنم اما می توانید از هر IDE یا CLI دیگری نیز استفاده کنید.
پس از موفقیت پروژه ، ایجاد شده باید وابستگی های زیر را به POM.xml خود اضافه کنیم.
org.apache.logging.log4j
log4j-api
${log4j2.version}
org.apache.logging.log4j
log4j-core
${log4j2.version}
org.apache.logging.log4j
log4j-slf4j-impl
${log4j2.version}
org.projectlombok
lombok
provided
1.18.30
org.apache.kafka
connect-runtime
3.7.2
وابستگی ها “log4j-api“،”دارای مدفوع“و”log4j-slf4j-impl“برای اهداف ورود به سیستم استفاده می شود. برای مشخص کردن فرمت سیاهههای خود ، ما باید Log4J2.xml را نیز به ClassPath خود اضافه کنیم.
نکته دیگر که در اینجا باید به آن توجه داشت ، وابستگی “اتصال-رنت” است. ما فقط باید فقط این وابستگی را اضافه کنیم و کتابخانه ها مانند “وصل کردن“،”اتصال دهنده“و”ترانسفورماتوناز آنجا که همه این کتابخانه ها نیز بخشی از ClassPath ما هستند ، برنامه Kafka Connect ما به عنوان یک سرور کار می کند و ما باید با استفاده از نقاط انتهایی REST بتوانیم با آن ارتباط برقرار کنیم.
ما همچنین باید اضافه کنیم “بهار-بوت-مرفین-پلوژین” به POM.xml ما در بخش ساخت. این به ما امکان می دهد تا شیشه چربی را تولید کنیم و کلاس را ارائه دهیم که روش اصلی دارد. این روش در شروع شیشه خوانده می شود.
org.springframework.boot
spring-boot-maven-plugin
dev.ankis.KafkaConnectApplication
repackage
3.4.2
به عنوان آخرین مرحله ، ما باید یک فایل پیکربندی را با نام “توزیع شده-config.properties” تحت SRC/Main/Resources اضافه کنیم. این تمام خصوصیات پیکربندی را ذخیره می کند.
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
listeners=HTTP://:8083
این تنظیمات را ارسال کنید ، می توانیم پیش برویم و برنامه را اجرا کنیم.
برای دیدن اینکه آیا سرور کانکتور Kafka ما در حال اجرا است ، می توانیم حلقه زیر را اجرا کنیم
curl --request GET \
--url http://localhost:8083/connector-plugins
منابع
