![](\"https:\/\/miro.medium.com\/v2\/resize:fit:1050\/1*Lg2m8BY3PCLTCkLpoSzxVQ.png\")
<\/p>\n<\/p>\n
Merhabalar bu yaz\u0131mda sizlere Kafka<\/strong>‘da bulunan verilerimizi nas\u0131l Kafka Connector<\/strong> yard\u0131m\u0131 ile HDFS<\/strong>‘e ta\u015f\u0131yabilece\u011fimizi g\u00f6sterece\u011fim. \u0130lk \u00f6nce gerekli Connector<\/strong>‘umuzu indirelim ve servislerimizi ba\u015flatal\u0131m. <\/p>\n # HDFS3 Sink Connector indirme i\u015flemi<\/span><\/strong> # Gerekli servisleri ba\u015flatma<\/span><\/strong> # Servislerin durumunu takip etmek i\u00e7in <\/span><\/strong> # Kontrol Merkezi yerine kullanaca\u011f\u0131m AKHQ servisini ba\u015flatmak i\u00e7in<\/span><\/strong> # akhq.yml i\u00e7eri\u011fi <\/span><\/strong> AKHQ Web UI<\/strong> eri\u015fimi i\u00e7in: http:\/\/localhost:8880\/ui\/<\/p><\/blockquote>\n Gerekli servislerimizi ba\u015flatt\u0131k ve \u00e7al\u0131\u015fma durumlar\u0131n\u0131 kontrol ettikten sonra \u015fimdi ise Kafka<\/strong>‘da berlitti\u011fimiz topic’e veri aktaracak olan Python<\/strong> kodumuza bakal\u0131m.<\/p>\n Yukar\u0131daki kod’u \u00f6zetlemek gerekirse yapaca\u011f\u0131 i\u015f rastgele bir \u015fekilde sipari\u015f verisi yaratarak bunu avro format\u0131nda Kafka<\/strong>‘da istedi\u011fimiz topic’e iletmektir. \u00d6rnek \u00e7\u0131kt\u0131 a\u015fa\u011f\u0131daki gibidir: <\/p>\n {‘CustomerID’: 824, ‘CustomerName’: ‘Robert Bullock’, ‘Email’: ‘robert.bullock@gmail.com’, ‘Address’: ‘9618 Chambers Centers Apt. 343 Kevinmouth, WA 39994’, ‘State’: ‘WA’, ‘PhoneNumber’: ‘399.352.69’, ‘InvoiceDate’: ‘2023-06-04 21:44:54’, ‘StockCode’: ‘23371’, ‘Description’: ‘SET 36 COLOUR PENCILS SPACEBOY’, ‘UnitPrice’: 2.46, ‘Quantity’: 1} <\/p>\n<\/blockquote>\n Kafka<\/strong>‘ya iletilen verileri Kafka CLI <\/strong>yard\u0131m\u0131 ile de g\u00f6r\u00fcnt\u00fclemek istersek a\u015fa\u011f\u0131daki script’i kullanabiliriz.<\/p>\n kafka-avro-console-consumer \\ Python<\/strong> kodunu durdurmad\u0131\u011f\u0131n\u0131z s\u00fcrece s\u00fcrekli sipari\u015f verisi \u00fcretilip istedi\u011fimiz konu’e iletilecek. \u015eimdi gelin bu verileri parke<\/strong> format\u0131nda HDFS<\/strong>‘e yazmam\u0131z\u0131 sa\u011fl\u0131yacak ba\u011flay\u0131c\u0131 konfig\u00fcrasyonunu<\/strong> olu\u015fturup bunu \u00e7al\u0131\u015ft\u0131ral\u0131m. <\/p>\n Konfig\u00fcrasyonumuz yukar\u0131daki gibidir. \u0130lgili konekt\u00f6r’u \u00e7al\u0131\u015ft\u0131rmak i\u00e7in ben Postman<\/strong> arac\u0131n\u0131 kulland\u0131m siz CLI<\/strong> yard\u0131m\u0131yla da bunu yapabilirsiniz. AKHQ Web UI<\/strong> ekran\u0131na giderek ba\u015far\u0131l\u0131 bir \u015fekilde \u00e7al\u0131\u015f\u0131p \u00e7al\u0131\u015fmama durumunu takip edebiliriz.<\/p>\n Sa\u011fl\u0131kl\u0131 bir \u015fekilde \u00e7al\u0131\u015ft\u0131\u011f\u0131n\u0131 g\u00f6rd\u00fck. \u015eimdi son olarak HDFS<\/strong> taraf\u0131ndaki kontrol\u00fcm\u00fcz\u00fc yaparak senaryomuzu tamamlam\u0131\u015f olaca\u011f\u0131z. <\/p>\n HDFS<\/strong> taraf\u0131na bakt\u0131\u011f\u0131m\u0131zda istedi\u011fimiz gibi State kolonunu dikkate alarak verileri ay\u0131rm\u0131\u015f ve parke<\/strong> format\u0131nda kaydetmi\u015f.<\/p>\n <\/p>\n Umar\u0131m faydal\u0131 bir yaz\u0131 olmu\u015ftur. Sorular\u0131n\u0131z ve \u00f6nerileriniz i\u00e7in benimle ileti\u015fime ge\u00e7ebilirsiniz. Te\u015fekk\u00fcrler. <\/p>\n","protected":false},"excerpt":{"rendered":" Kafka Connect ile HDFS’e Veri G\u00f6nderme \u0130\u015flemi Merhabalar bu yaz\u0131mda sizlere Kafka‘da bulunan verilerimizi nas\u0131l Kafka Connector yard\u0131m\u0131 ile HDFS‘e ta\u015f\u0131yabilece\u011fimizi g\u00f6sterece\u011fim. […]<\/p>\n","protected":false},"author":1,"featured_media":5923,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"inline_featured_image":false,"footnotes":""},"categories":[87],"tags":[],"class_list":["post-6109","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-blog"],"yoast_head":"\n
\nconfluent-hub install confluentinc\/kafka-connect-hdfs3:1.1.25<\/p>\n
\nconfluent local<\/span> services zookeeper start
\nconfluent local services kafka start
\nconfluent local<\/span> services schema-registry start
\nconfluent local<\/span> services connect start<\/p>\n
\nkonfluent yerel<\/span> hi\u0307zmetler stat\u00fcs\u00fc<\/p>\n
\njava -Dmicronaut.config.files=akhq.yml -jar akhq.jar<\/p><\/blockquote>\n
\nakhq:<\/span>
\n ba\u011flant\u0131lar:<\/span>
\n Yerel:<\/span>
\n \u00d6zellikler:<\/span>
\n bootstrap.servers:<\/span> “localhost:9092”<\/span>
\n \u015fema-kay\u0131t defteri:<\/span>
\n url:<\/span> “http:\/\/localhost:8881”<\/span>
\n Ba\u011flan:<\/span>
\n –<\/span> isim:<\/span> “ba\u011flan”<\/span>
\n url:<\/span> “http:\/\/localhost:8083”<\/span>
\nMikronot:<\/span>
\n Sunucu:<\/span>
\n Liman:<\/span> 8880<\/span><\/p>\n\n
import pandas as pd\nfrom faker import Faker\nimport random as rd\nimport time\nfrom confluent_kafka import avro\nfrom confluent_kafka.admin import AdminClient, NewTopic\nfrom confluent_kafka.avro import AvroProducer,CachedSchemaRegistryClient\nimport re\nimport datetime\n\nSCHEMA_REGISTRY_URL = \"http:\/\/localhost:8881\"\nBROKER_URL = \"localhost:9092\"\n\nadmin_client = AdminClient({'bootstrap.servers': BROKER_URL})\norder_topic = [NewTopic('order-topic', num_partitions=1, replication_factor=1)]\n\nfs = admin_client.create_topics(order_topic)\n\nschema = avro.loads(\n \"\"\"{\n \"type\":\"record\",\n \"name\":\"myrecord\",\n \"fields\": [\n {\n \"name\": \"CustomerID\",\n \"type\": \"int\"\n },\n {\n \"name\": \"CustomerName\",\n \"type\": \"string\"\n },\n {\n \"name\": \"Email\",\n \"type\": \"string\"\n },\n {\n \"name\": \"Address\",\n \"type\": \"string\"\n },\n {\n \"name\": \"State\",\n \"type\": \"string\"\n },\n {\n \"name\": \"PhoneNumber\",\n \"type\": \"string\"\n },\n {\n \"name\": \"InvoiceDate\",\n \"type\": \"string\"\n },\n {\n \"name\": \"StockCode\",\n \"type\": \"string\"\n },\n {\n \"name\": \"Description\",\n \"type\": \"string\"\n },\n {\n \"name\": \"UnitPrice\",\n \"type\": \"float\"\n },\n {\n \"name\": \"Quantity\",\n \"type\": \"int\"\n }\n ]\n}\"\"\"\n )\n\nschema_registry = CachedSchemaRegistryClient({\"url\": SCHEMA_REGISTRY_URL})\n\ndata = pd.read_csv('OnlineRetail.csv', sep = ';')\n\ndata[\"InvoiceNo\"] = data[\"InvoiceNo\"].str.replace('A','')\ndata[\"InvoiceNo\"] = data[\"InvoiceNo\"].str.replace('C','')\ndata[\"InvoiceNo\"] = data[\"InvoiceNo\"].astype(\"int64\")\ndata[\"StockCode\"] = data[\"StockCode\"].astype(\"string\")\ndata[\"Description\"] = data[\"Description\"].astype(\"string\")\ndata[\"UnitPrice\"] = data[\"UnitPrice\"].str.replace(',','.').astype(\"float\")\n\ncols = ['StockCode','Description', 'UnitPrice']\nStock_Desc_Unit_lst = data[cols].values.tolist()\n\nfake = Faker()\n\ndef create_data(x):\n project_data = {}\n for i in range(x):\n project_data[i] = {}\n project_data[i]['CustomerID'] = i + 1\n project_data[i]['Name'] = fake.name()\n\n return project_data\n\ndf = pd.DataFrame(create_data(5000)).transpose()\n\ncols = ['CustomerID','Name']\nCustomerID_Name = df[cols].values.tolist()\nQ_df = data[data[\"Quantity\"]>0]\nQuantity = Q_df[\"Quantity\"].values.tolist()\n\nnow = datetime.datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")\n\ndef produce_kafka_dict(topic_name,json_as_dict):\n producer=AvroProducer({\"bootstrap.servers\": BROKER_URL}, schema_registry=schema_registry)\n\n producer.produce( topic=topic_name,\n value=json_as_dict,\n value_schema=schema)\n producer.flush()\n\nwhile True:\n order={}\n n = rd.randint(1,10)\n time.sleep(rd.randint(1,15))\n CID_Name = rd.choice(CustomerID_Name)\n order[\"CustomerID\"]= CID_Name[0]\n order[\"CustomerName\"] = CID_Name[1]\n order[\"Email\"] = CID_Name[1].lower().replace(' ','.') + \"@gmail.com\"\n order[\"Address\"] = fake.address().replace('\\n',' ')\n order[\"State\"] = re.findall(\"[A-Z]{2}\", order[\"Address\"])[0]\n order[\"PhoneNumber\"] = fake.phone_number()[0:10]\n order[\"InvoiceDate\"] = datetime.datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")\n for a in range(n):\n Stock_Desc_Unit = rd.choice(Stock_Desc_Unit_lst)\n order[\"StockCode\"] = Stock_Desc_Unit[0]\n order[\"Description\"] = Stock_Desc_Unit[1]\n order[\"UnitPrice\"] = Stock_Desc_Unit[2]\n order[\"Quantity\"] = rd.choice(Quantity)\n produce_kafka_dict('order-topic',order)\n print(order) \n<\/pre>\n<\/blockquote>\n\n
\n
\n–bootstrap-server localhost:9092 \\
\n–topic order-topic \\
\n–property schema.registry.url=http:\/\/localhost:8881<\/p>\n<\/blockquote>\n![\"\"](\"https:\/\/bentego.com\/wp-content\/uploads\/2023\/06\/1_Fwvtof0D17-SnurS6SZRdg.webp\")
<\/p>\n\n
{\n \"name\": \"hdfs3-sink-connector\",\n \"config\": {\n \"connector.class\": \"io.confluent.connect.hdfs3.Hdfs3SinkConnector\",\n \"tasks.max\": \"1\",\n \"topics\": \"order-topic\",\n \"hdfs.url\": \"hdfs:\/\/localhost:9000\",\n \"flush.size\": \"3\",\n \"key.converter\": \"org.apache.kafka.connect.storage.StringConverter\",\n \"value.converter\": \"io.confluent.connect.avro.AvroConverter\",\n \"value.converter.schema.registry.url\":\"http:\/\/localhost:8881\",\n \"confluent.topic.bootstrap.servers\": \"localhost:9092\",\n \"confluent.topic.replication.factor\": \"1\",\n \"format.class\":\"io.confluent.connect.hdfs3.parquet.ParquetFormat\",\n \"partitioner.class\":\"io.confluent.connect.storage.partitioner.FieldPartitioner\",\n \"partition.field.name\":\"State\"\n }\n}\n<\/pre>\n<\/blockquote>\n
\n![\"\"](\"https:\/\/bentego.com\/wp-content\/uploads\/2023\/06\/1_otq08-0MjvbTbBco5DyP9g.webp\")
<\/p>\n![\"\"](\"https:\/\/bentego.com\/wp-content\/uploads\/2023\/06\/1_fd5sI7FMZKU7qo4FOn0grg.webp\")
<\/p>\n\n
hdfs dfs -ls \/topics\/order-topic\/\n<\/pre>\n<\/blockquote>\n
![\"\"](\"https:\/\/bentego.com\/wp-content\/uploads\/2023\/06\/01.webp\")
<\/p>\n![\"\"](\"https:\/\/bentego.com\/wp-content\/uploads\/2023\/06\/02.webp\")
<\/p>\n