با استفاده از logstash برای تقسیم داده ها و ارسال آن به چندین خروجی

LogStash یک منبع باز منبع باز و پردازش داده های سمت سرور است که داده ها را می دهد ، آن را تبدیل می کند و سپس آن را به یک یا چند خروجی ارسال می کند. در این وبلاگ ، نمونه ای را ارائه می دهم که نشان می دهد نحوه استفاده از LogStash برای استفاده از داده های مختلف از بازار سهام و ارسال داده های مربوط به هر بازار سهام منحصر به فرد به یک خروجی مشخص. این با اجرای مراحل زیر انجام می شود:

1. نسخه هایی از هر سند را از یک جریان ورودی بورس ایجاد کنید.
2. هر نسخه را فیلتر کنید تا فقط شامل زمینه هایی باشد که برای یک بورس اوراق بهادار معتبر هستند
3. ابرداده را به هر نسخه اضافه کنید تا نشان دهد که داده های بازار سهام موجود در آن است.
4. ابرداده را در هر سند ارزیابی کنید تا سند را به خروجی صحیح هدایت کنید.

توجه داشته باشید که در این پست وبلاگ ، من از ارتباطات خط لوله به خط لوله (بتا از 6. 5) استفاده نمی کنم که احتمالاً می تواند به برخی از عملکردهای شرح داده شده در اینجا دست یابد.

پرونده ورودی مثال

ما به عنوان ورودی به LogStash ، از یک فایل CSV استفاده می کنیم که حاوی مقادیر معیار بورس سهام است. چند نمونه ورودی CSV در زیر آورده شده است:

1483230600،1628. 75،1678. 1772. 8،2443. 6 1483232400،1613. 63،1688. 5،1750. 5،2460. 2 1483234200،1606. 51،1678. 6،1718. 6،171718،1718،1718،1718،1718،1718،1718،1718،1718. 2 4. 1،1708. 1،2470. 4

مقادیر جدا شده کاما "زمان" و ارزش معیارهای بورس زیر را نشان می دهد: "DAX" ، "SMI" ، "CAC" و "FTSE". خطوط فوق را در یک فایل CSV به نام "Stocks. CSV" کپی و چسبانده اید تا از این به عنوان ورودی در خط لوله LogStash استفاده کنید.

مثال خط لوله logstash

در زیر ما یک خط لوله logstash ارائه خواهیم داد که موارد زیر را انجام می دهد:

1. مقادیر بازار سهام را به عنوان ورودی با فرمت CSV از یک فایل CSV بخوانید.
2. نقشه هر ردیف ورودی CSV را به یک سند JSON ، که در آن ستون های CSV به قسمتهای زیر JSON نقشه می کشند: "زمان" ، "DAX" ، "SMI" ، "CAC" و "FTSE" نقشه برداری کنید.
3. قسمت زمانی را به قالب یونیکس تبدیل کنید.
4. برای ایجاد دو نسخه از هر سند از افزونه کلون فیلتر استفاده کنید (این نسخه ها علاوه بر سند اصلی هستند). فیلتر کلون به طور خودکار یک قسمت "نوع" جدید را به هر کپی سند جدید اضافه می کند ، جایی که "نوع" با نامهای آورده شده در آرایه کلون ها مطابقت دارد. ما انواع مختلفی را به عنوان "clone_for_smi" یا "clone_for_ftse" تعریف کرده ایم و هر کلون در نهایت فقط داده های مربوط به بازار سهام "SMI" یا "FTSE" را شامل می شود.
5. برای هر کلون:
   1. از افزونه فیلتر هرس استفاده کنید تا تمام زمینه ها را حذف کنید به جز آن دسته از زمینه هایی که برای بازار سهام خاص لیست شده اند.
   2. ابرداده را به هر سند مربوط به "نوع" اضافه شده توسط تابع کلون اضافه کنید. این امر ضروری است زیرا ما از تابع prune استفاده می کنیم که "نوع" را که توسط تابع کلون درج شده است حذف می کند و این اطلاعات در مرحله خروجی برای هدایت سند به خروجی صحیح مورد نیاز است.

   در زیر یک خط لوله Logstash است که مراحل بالا را اجرا می کند (با شماره مراحل مربوطه به عنوان نظرات اضافه شده است). این خط لوله را برای اجرا در فایلی به نام "clones. conf" کپی کنید:

   ## ورودی مرحله 1"$/stocks.csv" # The following will ensure re-reading of full input # each time Logstash executes (useful for debugging). start_position => "beginning" sincedb_path => "/dev/null">>## مرحله 2 فیلتر["time","DAX","SMI","CAC","FTSE"] separator => "," convert => <'DAX' =>'float' 'SMI' => 'float' 'CAC' => 'float' 'FTSE' => 'float'>>## تاریخ مرحله 3['time', 'UNIX']>## مرحله 4 # خط زیر 2 # نسخه اضافی از هر سند ایجاد می کند (یعنی شامل # اصلی، در مجموع 3).# هر کپی به طور خودکار دارای یک فیلد "نوع" است که مطابق با نام داده شده در آرایه است. شبیه['clone_for_SMI', 'clone_for_FTSE']>## STEP 5 if [type] == 'clone_for_SMI'<# Remove everything except "SMI" prune [ "SMI"]>جهش پیدا کند <"[@metadata][type]" =>"only_SMI">>>else if [type] == 'clone_for_FTSE'[ "FTSE"]>جهش پیدا کند <"[@metadata][type]" =>"only_FTSE">>>>خروجی ## مرحله 6<# The following output to stdout is just for debugging # and can be removed stdout یاقوتیtrue>>if [@metadata][type] == 'only_SMI'"smi_data">>other if [@metadata][type] == 'only_FTSE'"ftse_data">>دیگر"stocks_original">>>

   آزمایش خط لوله Logstash

   برای آزمایش این خط لوله با داده های مثال CSV، می توانید دستور زیر را اجرا کنید و آن را تغییر دهید تا مطمئن شوید که از مسیرهایی که برای سیستم خود درست هستند استفاده می کنید. توجه داشته باشید که تعیین "config. reload. automatic" اختیاری است، اما به ما اجازه می دهد تا بدون راه اندازی مجدد Logstash، "clones. conf" را به طور خودکار بارگیری کنیم:

   ./logstas h-f ./clones. conf --config. reload. automatic

   هنگامی که Logstash فایل "stocks. csv" را خواند و پردازش را کامل کرد، می توانیم سه نمایه به نام های "smi_data"، "ftse_data" و "stocks_original" را مشاهده کنیم.

   شاخص SMI را بررسی کنید

   دریافت /smi_data/_search

   با این کار اسنادی با ساختار زیر نمایش داده می شود. توجه داشته باشید که فقط داده های "SMI" در فهرست "smi_data" ظاهر می شود.

   شاخص FTSE را بررسی کنید

   دریافت /ftse_data/_search

   با این کار اسنادی با ساختار زیر نمایش داده می شود. توجه داشته باشید که فقط فیلد "FTSE" در اسناد در فهرست "ftse_data" ظاهر می شود.

   فهرست اسناد اصلی را بررسی کنید

   دریافت /stocks_originals/_search

   با این کار اسنادی با ساختار زیر نمایش داده می شود. توجه داشته باشید که نسخه اصلی بدون فیلتر اسناد در فهرست "stocks_original" ظاهر می شود.

   نتیجه

   در این پست وبلاگ زیر مجموعه کوچکی از قابلیت های Logstash را نشان داده ام. به طور خاص، مثالی ارائه کردم که نشان می داد چگونه از Logstash برای دریافت داده ها از چندین بازار سهام، و سپس پردازش و هدایت آن داده ها به خروجی های مجزا استفاده کنیم. اگر Logstash و Elastic Stack را آزمایش می کنید و سؤالی دارید، در انجمن های بحث عمومی ما کمک بگیرید.