Apache NIFI - ප්‍රායෝගිකව ඇති අවස්ථා පිළිබඳ කෙටි දළ විශ්ලේෂණයක්

හැඳින්වීම

මගේ වර්තමාන සේවා ස්ථානයේ දී මට මෙම තාක්ෂණය පිළිබඳව දැන හඳුනා ගැනීමට සිදු විය. මම පොඩි පසුබිමකින් පටන් ගන්නම්. ඊළඟ රැස්වීමේදී, අපේ කණ්ඩායමට කිව්වා අපි සමඟ ඒකාබද්ධතාවයක් ඇති කර ගත යුතු බව දන්නා පද්ධතිය. ඒකාබද්ධ කිරීම මගින් අදහස් කළේ මෙම සුප්‍රසිද්ධ පද්ධතිය HTTP හරහා අපට ඉල්ලීම් නිශ්චිත අවසාන ලක්ෂ්‍යයකට එවන බවත්, අප අමුතු ලෙස ප්‍රතිචාර SOAP පණිවිඩයක් ආකාරයෙන් ආපසු එවන බවත් ය. සෑම දෙයක්ම සරල හා සුළු වශයෙන් පෙනේ. මෙයින් ඔබට අවශ්‍ය වන්නේ...

අරමුණු

සේවා 3 ක් සාදන්න. ඒවායින් පළමුවැන්න දත්ත සමුදා යාවත්කාලීන සේවාවයි. මෙම සේවාව, තෙවන පාර්ශවීය පද්ධතියකින් නව දත්ත පැමිණෙන විට, දත්ත සමුදායේ දත්ත යාවත්කාලීන කර එය ඊළඟ පද්ධතියට මාරු කිරීම සඳහා CSV ආකෘතියෙන් ගොනුවක් ජනනය කරයි. දෙවන සේවාවේ අවසාන ලක්ෂ්‍යය ලෙස හැඳින්වේ - මාරු කරන ලද ගොනුව ලබා ගන්නා FTP ප්‍රවාහන සේවය, එය වලංගු කර FTP හරහා ගොනු ගබඩාවේ තබයි. තෙවන සේවාව, පාරිභෝගික දත්ත හුවමාරු සේවාව, පළමු දෙක සමඟ අසමමිතිකව ක්‍රියා කරයි. එය ඉහත සාකච්ඡා කළ ගොනුව ලබා ගැනීම සඳහා තෙවන පාර්ශවීය බාහිර පද්ධතියකින් ඉල්ලීමක් ලබා ගනී, සූදානම් ප්‍රතිචාර ගොනුව ගෙන, එය වෙනස් කර (ID, විස්තරය, linkToFile ක්ෂේත්‍ර යාවත්කාලීන කරයි) සහ ප්‍රතිචාරය SOAP පණිවිඩයක ආකාරයෙන් යවයි. එනම්, සමස්ත පින්තූරය පහත පරිදි වේ: පළමු සේවා දෙක ඔවුන්ගේ වැඩ ආරම්භ කරන්නේ යාවත්කාලීන කිරීම සඳහා දත්ත පැමිණි විට පමණි. තෙවන සේවාව නිරන්තරයෙන් ක්‍රියාත්මක වන්නේ තොරතුරු පාරිභෝගිකයින් බොහෝමයක් සිටින නිසා, විනාඩියකට දත්ත සඳහා ඉල්ලීම් 1000 ක් පමණ වේ. සේවාවන් නිරන්තරයෙන් ලබා ගත හැකි අතර ඒවායේ අවස්ථා පරීක්ෂණ, ආදර්ශන, පූර්ව නිෂ්පාදන සහ නිෂ්පාදන වැනි විවිධ පරිසරවල පිහිටා ඇත. මෙම සේවාවන් ක්‍රියාත්මක වන ආකාරය දැක්වෙන රූප සටහනක් පහත දැක්වේ. අනවශ්‍ය සංකීර්ණතාවයන් මඟහරවා ගැනීම සඳහා සමහර විස්තර සරල කර ඇති බව වහාම පැහැදිලි කිරීමට මට ඉඩ දෙන්න.

තාක්ෂණික ගැඹුරු කිරීම

ගැටලුවට විසඳුමක් සැලසුම් කිරීමේදී, අපි මුලින්ම තීරණය කළේ Spring Framework, Nginx balancer, Postgres දත්ත සමුදාය සහ වෙනත් තාක්‍ෂණික හා තාක්‍ෂණික නොවන දේවල් භාවිතා කර ජාවා හි යෙදුම් සෑදීමට ය. තාක්ෂණික විසඳුමක් සංවර්ධනය කිරීමේ කාලය මෙම ගැටළුව විසඳීම සඳහා වෙනත් ප්‍රවේශයන් සලකා බැලීමට අපට ඉඩ සලසා දුන් බැවින්, අපගේ බැල්ම ඇතැම් කවයන් තුළ විලාසිතාවක් වන Apache NIFI තාක්‍ෂණය වෙත යොමු විය. මෙම තාක්ෂණය අපට මෙම සේවාවන් 3 දැකීමට ඉඩ දුන් බව මම වහාම කියමි. මෙම ලිපියෙන් පාරිභෝගිකයාට ගොනු ප්‍රවාහන සේවාවක් සහ දත්ත හුවමාරු සේවාවක් සංවර්ධනය කිරීම විස්තර කරනු ඇත, නමුත් ලිපිය ප්‍රයෝජනවත් නම්, දත්ත සමුදායේ දත්ත යාවත්කාලීන කිරීම සඳහා වන සේවාව ගැන මම ලියන්නෙමි.

එය කුමක්ද

NIFI යනු වේගයෙන් සමාන්තර පැටවීම සහ දත්ත සැකසීම සඳහා බෙදා හරින ලද ගෘහ නිර්මාණ ශිල්පයකි, මූලාශ්‍ර සහ පරිවර්තනයන් සඳහා ප්ලගීන විශාල ප්‍රමාණයක්, වින්‍යාසය අනුවාදනය කිරීම සහ තවත් බොහෝ දේ. හොඳ ප්‍රසාද දීමනාවක් නම් එය භාවිතා කිරීමට ඉතා පහසු වීමයි. getFile, sendHttpRequest සහ වෙනත් සුළු ක්‍රියාවලීන් වර්ග ලෙස නිරූපණය කළ හැක. සෑම චතුරශ්‍රයක්ම ක්‍රියාවලියක් නියෝජනය කරයි, එහි අන්තර්ක්‍රියා පහත රූපයේ දැකිය හැකිය. ක්‍රියාවලි සැකසුම් අන්තර්ක්‍රියා පිළිබඳ වඩාත් සවිස්තරාත්මක ලේඛන ලියා ඇත මෙහි , රුසියානු භාෂාව කතා කරන අය සඳහා - මෙහි. ප්‍රලේඛනය NIFI ඉවත් කර ධාවනය කරන්නේ කෙසේද යන්න මෙන්ම වර්ග ලෙසද හැඳින්වෙන ක්‍රියාවලි නිර්මාණය කරන්නේ කෙසේද යන්න මනාව විස්තර කරයි
ලිපියක් ලිවීමේ අදහස උපත ලැබුවේ දිගු සෙවුමකින් සහ ලැබුණු තොරතුරු සවිඥානක දෙයකට ව්‍යුහගත කිරීමෙන් මෙන්ම අනාගත සංවර්ධකයින්ට ජීවිතය ටිකක් පහසු කිරීමට ඇති ආශාවෙනි.

උදාහරණ:

චතුරස්රයන් එකිනෙකා සමඟ අන්තර් ක්රියා කරන ආකාරය පිළිබඳ උදාහරණයක් සලකා බලයි. සාමාන්‍ය යෝජනා ක්‍රමය තරමක් සරල ය: අපට HTTP ඉල්ලීමක් ලැබේ (න්‍යායාත්මකව, ඉල්ලීමේ අන්තර්ගතයේ ගොනුවක් සමඟ. NIFI හි හැකියාවන් ප්‍රදර්ශනය කිරීම සඳහා, මෙම උදාහරණයේදී ඉල්ලීම දේශීය ගොනු ගබඩාවෙන් ගොනුවක් ලබා ගැනීමේ ක්‍රියාවලිය ආරම්භ කරයි. ), එවිට අපි FH වෙතින් ගොනුවක් ලබා ගැනීමේ ක්‍රියාවලියට සමාන්තරව FTP හරහා FH වෙත ගෙන යාමේ ක්‍රියාවලියට සමාන්තරව ඉල්ලීම ලැබී ඇති බවට ප්‍රතිචාරයක් යවන්නෙමු. ක්රියාවලීන් ඊනියා ප්රවාහ ගොනුව හරහා එකිනෙකා සමඟ අන්තර් ක්රියා කරන බව පැහැදිලි කිරීම වටී. NIFI හි ගුණාංග සහ අන්තර්ගතය ගබඩා කරන මූලික ආයතනය මෙයයි. අන්තර්ගතය යනු ප්‍රවාහ ගොනුව මගින් නිරූපණය වන දත්ත වේ. එනම්, දළ වශයෙන් කිවහොත්, ඔබ එක් චතුරස්රයකින් ගොනුවක් ලබා ගෙන එය තවත් කොටසකට මාරු කළහොත්, අන්තර්ගතය ඔබේ ගොනුව වනු ඇත.

ඔබට පෙනෙන පරිදි, මෙම පින්තූරය සාමාන්ය ක්රියාවලිය පෙන්නුම් කරයි. HandleHttpRequest - ඉල්ලීම් පිළිගනී, ReplaceText - ප්‍රතිචාර ශරීරයක් ජනනය කරයි, HandleHttpResponse - ප්‍රතිචාරයක් යවයි. FetchFile - ගොනු ගබඩාවකින් ගොනුවක් ලබා ගනී, එය PutSftp චතුරස්‍රයට මාරු කරයි - මෙම ගොනුව FTP මත, නිශ්චිත ලිපිනයෙහි තබයි. දැන් මෙම ක්රියාවලිය ගැන වැඩි විස්තර.

මෙම අවස්ථාවේ දී, ඉල්ලීම සෑම දෙයකම ආරම්භයයි. එහි වින්‍යාස පරාමිතීන් දෙස බලමු.

StandardHttpContextMap හැර මෙහි ඇති සෑම දෙයක්ම ඉතා සුළු දෙයකි - මෙය ඔබට ඉල්ලීම් යැවීමට සහ ලැබීමට ඉඩ සලසන ආකාරයේ සේවාවකි. වඩාත් විස්තරාත්මකව සහ උදාහරණ සමඟ පවා ඔබට දැකිය හැකිය - මෙහි

ඊළඟට, අපි චතුරස්රයේ ReplaceText වින්යාස පරාමිතීන් දෙස බලමු. ReplacementValue වෙත අවධානය යොමු කිරීම වටී - මෙය ප්‍රතිචාරයක ස්වරූපයෙන් පරිශීලකයාට ආපසු ලබා දෙනු ඇත. සැකසීම් තුළ ඔබට ලොග් වීමේ මට්ටම වෙනස් කළ හැකිය, ඔබට ලොග් {ඔබ nifi-1.9.2/logs ඉවත් කළ ස්ථානයේ}/nifi-XNUMX/logs දැකිය හැකිය, අසාර්ථක/සාර්ථක පරාමිතීන් ද ඇත - මෙම පරාමිති මත පදනම්ව ඔබට ක්‍රියාවලිය සමස්තයක් ලෙස නියාමනය කළ හැකිය. . එනම්, සාර්ථක පෙළ සැකසීමේදී, පරිශීලකයාට ප්‍රතිචාරයක් යැවීමේ ක්‍රියාවලිය කැඳවනු ලබන අතර, තවත් අවස්ථාවකදී අපි අසාර්ථක ක්‍රියාවලිය සරලව ලොග් කරන්නෙමු.

HandleHttpResponse ගුණාංගවල ප්‍රතිචාරයක් සාර්ථකව නිර්මාණය කළ විට තත්ත්වය හැර විශේෂ රසවත් කිසිවක් නොමැත.

අපි ඉල්ලීම සහ ප්‍රතිචාරය නිරාකරණය කර ඇත - අපි ගොනුව ලබාගෙන එය FTP සේවාදායකයේ තැබීමට යමු. FetchFile - සැකසුම් වල දක්වා ඇති මාර්ගයෙන් ගොනුවක් ලබාගෙන එය ඊළඟ ක්‍රියාවලියට යවයි.

ඉන්පසු PutSftp වර්ග - ගොනුව ගොනු ගබඩාවේ තබයි. අපට පහත වින්‍යාස පරාමිතීන් දැකිය හැකිය.

එක් එක් චතුරශ්රය දියත් කළ යුතු වෙනම ක්රියාවලියක් බව අවධානය යොමු කිරීම වටී. කිසිදු සංකීර්ණ අභිරුචිකරණයක් අවශ්‍ය නොවන සරලම උදාහරණය අපි බැලුවෙමු. ඊළඟට, අපි ක්රියාවලිය ටිකක් සංකීර්ණ ලෙස බලමු, එහිදී අපි කට්ට මත ටිකක් ලියන්නෙමු.

වඩාත් සංකීර්ණ උදාහරණයක්

SOAP පණිවිඩය වෙනස් කිරීමේ ක්‍රියාවලිය හේතුවෙන් පාරිභෝගිකයාට දත්ත හුවමාරු සේවාව ටිකක් සංකීර්ණ විය. සාමාන්ය ක්රියාවලිය පහත රූපයේ දැක්වේ.

මෙහි අදහස ද විශේෂයෙන් සංකීර්ණ නොවේ: පාරිභෝගිකයාගෙන් ඔහුට දත්ත අවශ්‍ය බවට ඉල්ලීමක් ලැබී, ඔහුට පණිවිඩයක් ලැබුණු බවට ප්‍රතිචාරයක් යවා, ප්‍රතිචාර ගොනුව ලැබීමේ ක්‍රියාවලිය ආරම්භ කර, පසුව එය යම් තර්කයකින් සංස්කරණය කර, පසුව SOAP පණිවිඩයක ස්වරූපයෙන් පාරිභෝගිකයා වෙත ගොනුව සේවාදායකය වෙත මාරු කරන ලදී.

මම හිතන්නේ අපි ඉහත දුටු එම කොටු නැවත විස්තර කිරීමට අවශ්‍ය නැත - අපි කෙලින්ම නව ඒවාට යමු. ඔබට කිසියම් ගොනුවක් සංස්කරණය කිරීමට අවශ්‍ය නම් සහ සාමාන්‍ය ReplaceText වර්ගයේ වර්ග සුදුසු නොවේ නම්, ඔබට ඔබේම පිටපතක් ලිවීමට සිදුවේ. ExecuteGroogyScript චතුරස්රය භාවිතයෙන් මෙය කළ හැක. එහි සැකසුම් පහත දැක්වේ.

මෙම චතුරශ්‍රයට ස්ක්‍රිප්ට් පැටවීම සඳහා විකල්ප දෙකක් ඇත. පළමුවැන්න නම් ස්ක්‍රිප්ට් එකක් සහිත ගොනුවක් බාගැනීමයි. දෙවැන්න scriptBody වෙත ස්ක්‍රිප්ට් ඇතුල් කිරීමෙනි. මා දන්නා පරිදි, executeScript චතුරශ්‍රය භාෂා කිහිපයකට සහය දක්වයි - ඒවායින් එකක් ග්‍රෝවි ය. මම ජාවා සංවර්ධකයින් බලාපොරොත්තු සුන් කරමි - ඔබට එවැනි වර්ගවල ජාවා හි ස්ක්‍රිප්ට් ලිවිය නොහැක. ඇත්ත වශයෙන්ම අවශ්ය අය සඳහා, ඔබ ඔබේම අභිරුචි චතුරස්රයක් සාදා එය NIFI පද්ධතියට එක් කළ යුතුය. මෙම සම්පූර්ණ මෙහෙයුම රබන් සමඟ තරමක් දිගු නර්තනයක් සමඟ ඇත, එය අපි මෙම ලිපියෙන් ගනුදෙනු නොකරමු. මම තෝරගත්තෙ විකාර භාෂාව. පහත දැක්වෙන්නේ SOAP පණිවිඩයක හැඳුනුම්පත වැඩි දියුණු කරන පරීක්ෂණ පිටපතකි. එය සැලකිල්ලට ගැනීම වැදගත්ය. ඔබ flowFile වෙතින් ගොනුව ගෙන එය යාවත්කාලීන කරන්න, ඔබට එය නැවත එහි තැබිය යුතු බව අමතක නොකරන්න, යාවත්කාලීන කර ඇත. සියලුම පුස්තකාල ඇතුළත් නොවන බව ද සඳහන් කිරීම වටී. ඔබට තවමත් ලිබ් එකක් ආනයනය කිරීමට සිදු විය හැකිය. තවත් අවාසියක් නම් මෙම චතුරශ්‍රයේ ඇති ස්ක්‍රිප්ටය දෝෂහරණය කිරීම තරමක් අපහසු වීමයි. NIFI JVM වෙත සම්බන්ධ වී නිදොස් කිරීමේ ක්‍රියාවලිය ආරම්භ කිරීමට ක්‍රමයක් තිබේ. පුද්ගලිකව, මම දේශීය යෙදුමක් දියත් කළ අතර සැසියෙන් ගොනුවක් ලැබීම අනුකරණය කළෙමි. මම දේශීයව ද දෝෂහරණය කළා. ස්ක්‍රිප්ට් පූරණය කිරීමේදී දිස්වන දෝෂ Google හට තරමක් පහසු වන අතර NIFI විසින්ම ලොගයට ලියා ඇත.

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

ඇත්ත වශයෙන්ම, චතුරස්රයේ අභිරුචිකරණය අවසන් වන්නේ මෙයයි. ඊළඟට, යාවත්කාලීන කරන ලද ගොනුව චතුරස්රයට මාරු කරනු ලැබේ, ගොනුව සේවාදායකයට යැවීම සඳහා වගකිව යුතුය. මෙම චතුරශ්‍රය සඳහා සැකසුම් පහත දැක්වේ.

SOAP පණිවිඩයක් සම්ප්‍රේෂණය වන ක්‍රමය අපි විස්තර කරමු. අපි ලියන්නේ කොහෙද කියලා. මීළඟට ඔබ මෙය SOAP බව සඳහන් කළ යුතුය.

සත්කාරක සහ ක්‍රියාව (soapAction) වැනි ගුණාංග කිහිපයක් එක් කරන්න. අපි සුරැකීමට සහ පරීක්ෂා කරන්න. SOAP ඉල්ලීම් යවන්නේ කෙසේද යන්න පිළිබඳ වැඩි විස්තර ඔබට දැක ගත හැක මෙහි

අපි NIFI ක්‍රියාවලි භාවිතා කිරීම සඳහා විකල්ප කිහිපයක් දෙස බැලුවෙමු. ඔවුන් අන්තර්ක්‍රියා කරන්නේ කෙසේද සහ ඔවුන්ගේ සැබෑ ප්‍රයෝජනය කුමක්ද? සලකා බැලූ උදාහරණ පරීක්ෂණ ඒවා වන අතර සටනේදී ඇත්ත වශයෙන්ම සිදු වන දෙයට වඩා තරමක් වෙනස් වේ. මෙම ලිපිය සංවර්ධකයින්ට ටිකක් ප්‍රයෝජනවත් වනු ඇතැයි මම බලාපොරොත්තු වෙමි. ඔබගේ අවදානය පිළිබඳ ස්තූතියි. ඔබට කිසියම් ප්රශ්නයක් ඇත්නම්, ලියන්න. මම උත්තර දෙන්න උත්සහ කරන්නම්.

මූලාශ්රය: www.habr.com