Apache NIFI - በተግባር ላይ ያሉ እድሎች አጭር መግለጫ

መግቢያ

አሁን ባለሁበት የስራ ቦታ ከዚህ ቴክኖሎጂ ጋር መተዋወቅ ነበረብኝ። በትንሽ ዳራ እጀምራለሁ. በሚቀጥለው ሰልፍ ቡድናችን ከ ጋር ውህደት መፍጠር እንዳለብን ተነግሮናል። የታወቀ ስርዓት. ውህደት ማለት ይህ በጣም የታወቀ ስርዓት ጥያቄዎችን በኤችቲቲፒ በኩል ወደ አንድ የተወሰነ የመጨረሻ ነጥብ ይልክልናል፣ እና በሚያስገርም ሁኔታ፣ ምላሾችን በሶፕ መልእክት መልክ እንመልሳለን። ሁሉም ነገር ቀላል እና ቀላል ይመስላል. ከዚህ ቀጥሎ ያለው…

ዓላማ

3 አገልግሎቶችን ይፍጠሩ. የመጀመሪያው የዳታቤዝ ማሻሻያ አገልግሎት ነው። ይህ አገልግሎት ከሶስተኛ ወገን አዲስ መረጃ ከተቀበለ በኋላ በመረጃ ቋቱ ውስጥ ያለውን መረጃ ያሻሽላል እና ወደ ቀጣዩ ስርዓት ለማስተላለፍ በCSV ቅርጸት ፋይል ያመነጫል። የሁለተኛው አገልግሎት የመጨረሻ ነጥብ ይባላል - በኤፍቲፒ በኩል ያለው የትራንስፖርት አገልግሎት የተላለፈውን ፋይል ይቀበላል ፣ ያፀድቃል እና በኤፍቲፒ በኩል በፋይል ማከማቻ ውስጥ ያደርገዋል። ሦስተኛው አገልግሎት - ለተጠቃሚው የውሂብ ማስተላለፍ አገልግሎት, ከመጀመሪያዎቹ ሁለት ጋር በማይመሳሰል መልኩ ይሰራል. ከላይ የተብራራውን ፋይል ለመቀበል ከሶስተኛ ወገን የውጭ ስርዓት ጥያቄ ይቀበላል ፣ ዝግጁ የሆነ የምላሽ ፋይል ወስዶ አሻሽሎታል (መታወቂያውን ፣ መግለጫውን ፣ አገናኝ ቶፋይል መስኮችን ያዘምናል) እና ምላሽ በ a መልክ ይልካል ። የሳሙና መልእክት። ያም በአጠቃላይ, ስዕሉ እንደሚከተለው ነው-የመጀመሪያዎቹ ሁለት አገልግሎቶች ሥራቸውን የሚጀምሩት ለማዘመን መረጃው ሲደርስ ብቻ ነው. ሶስተኛው አገልግሎት ያለማቋረጥ ይሰራል ምክንያቱም ብዙ የመረጃ ተጠቃሚዎች ስላሉ፣ በደቂቃ ወደ 1000 የሚጠጉ የመረጃ ጥያቄዎች። አገልግሎቶቹ ሁል ጊዜ ይገኛሉ እና ምሳሌዎቻቸው በተለያዩ አካባቢዎች ላይ ይገኛሉ፣ ለምሳሌ ሙከራ፣ ማሳያ፣ ፕሪፕሮድ እና ፕሮድ። ከታች እነዚህ አገልግሎቶች እንዴት እንደሚሠሩ የሚያሳይ ሥዕላዊ መግለጫ ነው. አንዳንድ ዝርዝሮች አላስፈላጊ ውስብስብ ነገሮችን ለማስወገድ ቀለል ያሉ መሆናቸውን ወዲያውኑ ላብራራ።

ቴክኒካዊ ጥልቀት

ለችግሩ መፍትሄ ለማቀድ ስናቅድ መጀመሪያ የፀደይ ማዕቀፉን ፣ የ Nginx ሚዛን ፣ የ Postgres ዳታቤዝ እና ሌሎች ቴክኒካዊ እና ቴክኒካዊ ያልሆኑ ነገሮችን በመጠቀም በጃቫ ውስጥ መተግበሪያዎችን ለመስራት ወሰንን ። ቴክኒካል መፍትሔ ለማዘጋጀት ጊዜ ይህን ችግር ለመፍታት ሌሎች አቀራረቦችን ከግምት አስችሏል ጀምሮ, መልክ Apache NIFI ቴክኖሎጂ ላይ ወደቀ, አንዳንድ ክበቦች ውስጥ ፋሽን. ይህ ቴክኖሎጂ እነዚህን 3 አገልግሎቶች እንድናስተውል እንደፈቀደልን ወዲያውኑ መናገር አለብኝ። ይህ ጽሑፍ የፋይል ማጓጓዣ አገልግሎትን እና ለተጠቃሚው የውሂብ ማስተላለፍ አገልግሎት እድገትን ይገልፃል, ነገር ግን ጽሑፉ ከገባ, በመረጃ ቋቱ ውስጥ ስላለው የውሂብ ማሻሻያ አገልግሎት እጽፋለሁ.

ይህ ምንድን ነው?

NIFI ለፈጣን ትይዩ ጭነት እና መረጃን ለማስኬድ የተከፋፈለ አርክቴክቸር ነው፣ ብዙ ቁጥር ያላቸው ፕለጊኖች ለመረጃ ምንጮች እና ለውጦች፣ የማዋቀር ስሪት እና ሌሎች ብዙ። ጥሩ ጉርሻ ለመጠቀም በጣም ቀላል ነው። እንደ getFile፣ sendHttpRequest እና ሌሎች ያሉ ጥቃቅን ሂደቶች እንደ ካሬዎች ሊወከሉ ይችላሉ። እያንዳንዱ ካሬ አንድ የተወሰነ ሂደትን ይወክላል, የእሱ መስተጋብር ከዚህ በታች ባለው ስእል ውስጥ ይታያል. በሂደቱ ማዋቀር መስተጋብር ላይ የበለጠ ዝርዝር ሰነዶች ተጽፈዋል እዚህ በሩሲያኛ ላሉ ሰዎች - እዚህ. ሰነዱ NIFI እንዴት እንደሚፈታ እና እንደሚያስኬድ እንዲሁም ሂደቶችን እንዴት መፍጠር እንደሚቻል በትክክል ይገልጻል።
ጽሑፍ የመጻፍ ሀሳብ ከረዥም ጊዜ ፍለጋ በኋላ የተወለደ እና የተቀበለውን መረጃ ወደ አንድ ንቃተ-ህሊና በማዋቀር እንዲሁም ህይወትን ለወደፊት ገንቢዎች ትንሽ ቀላል የማድረግ ፍላጎት ነው ።

ለምሳሌ:

ካሬዎች እርስ በእርሳቸው እንዴት እንደሚገናኙ የሚያሳይ ምሳሌ ግምት ውስጥ ይገባል. አጠቃላይ መርሃግብሩ በጣም ቀላል ነው የኤችቲቲፒ ጥያቄን እንቀበላለን (በንድፈ ሀሳብ, በጥያቄው አካል ውስጥ ካለው ፋይል ጋር. የ NIFI ችሎታዎችን ለማሳየት, በዚህ ምሳሌ, ጥያቄው ከአካባቢው ኤፍኤች ፋይል የማግኘት ሂደት ይጀምራል), ከዚያም ጥያቄው እንደደረሰው ምላሽ እንልካለን, በትይዩ, ከኤፍኤች ፋይል የማግኘት ሂደት እና ከዚያም በኤፍቲፒ ወደ ኤፍኤች የማዘዋወር ሂደት. ሂደቶቹ እርስ በእርሳቸው እንደሚገናኙ ማብራራት ጠቃሚ ነው ፍሰት ፋይል ተብሎ በሚጠራው. ይህ በNIFI ውስጥ ባህሪያትን እና ይዘቶችን የሚያከማች መሰረታዊ አካል ነው። ይዘት - በዥረት ፋይል የተወከለው ውሂብ። ማለትም፣ በግምት፣ ከአንድ ካሬ ፋይል ከተቀበሉ እና ወደ ሌላ ካስተላለፉ ፋይልዎ ይዘቱ ይሆናል።

እንደሚመለከቱት, ይህ ስዕል አጠቃላይ ሂደቱን ያሳያል. HandleHttpጥያቄ - ጥያቄዎችን ይቀበላል ፣ ጽሑፍን ይተኩ - ምላሽ ሰጪ አካል ያመነጫል ፣ HandleHttpResponse - ምላሽ ይሰጣል። FetchFile - ከፋይል ማከማቻው ፋይል ይቀበላል እና ወደ PutSftp ካሬ ያስተላልፋል - ይህንን ፋይል በኤፍቲፒ ላይ በተጠቀሰው አድራሻ ያስቀምጣል. አሁን ስለዚህ ሂደት የበለጠ።

በዚህ ጉዳይ ላይ ጥያቄ የሁሉም ነገር መጀመሪያ ነው። የእሱን የማዋቀር አማራጮችን እንይ.

ከStandardHttpContextMap በስተቀር ሁሉም ነገር እዚህ ጋር በጣም ቀላል ነው - ይህ ጥያቄን ለመላክ እና ለመቀበል የሚያስችል የአገልግሎት ዓይነት ነው። ለተጨማሪ ዝርዝሮች እና በምሳሌዎች እንኳን ማየት ይችላሉ - እዚህ

በመቀጠል፣ ReplaceText የካሬ ውቅር አማራጮችን እንመልከት። ለ ReplacementValue ትኩረት መስጠት ተገቢ ነው - ይህ ለተጠቃሚው እንደ ምላሽ የሚመለሰው ነው. በቅንብሮች ውስጥ የምዝግብ ማስታወሻውን ደረጃ ማስተካከል ይችላሉ ፣ ምዝግብ ማስታወሻዎቹን ማየት ይችላሉ {nifi ያልታሸገበት}/nifi-1.9.2/logs ፣ እንዲሁም ውድቀት / የስኬት መለኪያዎች አሉ - በእነዚህ መለኪያዎች ላይ በመመስረት ሂደቱን መቆጣጠር ይችላሉ ሙሉ። ያም ማለት የተሳካ የጽሁፍ ሂደትን በተመለከተ ለተጠቃሚው ምላሽ የመላክ ሂደት ይጠራል, እና በሌላ ሁኔታ, በቀላሉ ያልተሳካውን ሂደት ቃል እንገባለን.

ምላሹ በተሳካ ሁኔታ ከተፈጠረበት ሁኔታ በስተቀር በ HandleHttpResponse ባህሪያት ውስጥ ምንም የሚያስደስት ነገር የለም።

የጥያቄውን ምላሽ አግኝተናል - ወደ ፋይሉ መቀበል እና በኤፍቲፒ አገልጋይ ላይ እናስቀምጠው። FetchFile - በቅንብሮች ውስጥ ከተጠቀሰው መንገድ ፋይል ይቀበላል እና ወደሚቀጥለው ሂደት ያስተላልፋል።

እና ከዚያ ካሬው PutSftp - ፋይሉን በፋይል ማከማቻ ውስጥ ያስቀምጣል. የማዋቀሪያ አማራጮች ከዚህ በታች ሊታዩ ይችላሉ.

እያንዳንዱ ካሬ መጀመር ያለበት የተለየ ሂደት መሆኑን ትኩረት መስጠቱ ተገቢ ነው. ምንም ውስብስብ ማበጀት የማይፈልገውን በጣም ቀላሉ ምሳሌ ተመልክተናል. በመቀጠል, ሂደቱን ትንሽ ውስብስብ እናስባለን, እዚያም በሾለኞቹ ላይ ትንሽ እንጽፋለን.

የበለጠ ውስብስብ ምሳሌ

ለተጠቃሚው የውሂብ ማስተላለፍ አገልግሎት የሶፕ መልእክትን በማስተካከል ሂደት ትንሽ የተወሳሰበ ነው. አጠቃላይ ሂደቱ ከዚህ በታች ባለው ስእል ውስጥ ይታያል.

እዚህ ፣ ሀሳቡ በጣም የተወሳሰበ አይደለም-ከተጠቃሚው መረጃ እንደሚያስፈልገው ጥያቄ ደረሰን ፣ መልእክት እንደተቀበለ ምላሽ ልከናል ፣ የምላሽ ፋይልን የመቀበል ሂደት ጀመርን ፣ ከዚያም በተወሰነ አመክንዮ አስተካክለው እና ከዚያ አስተላልፈዋል። ፋይሉን ለተጠቃሚው በሶፕ መልእክት መልክ ለአገልጋዩ።

እኔ እንደማስበው እነዚያን ከላይ ያየናቸውን አደባባዮች እንደገና መግለጽ ጠቃሚ አይደለም - ወዲያውኑ ወደ አዳዲሶቹ እንሂድ። ማንኛውንም ፋይል ማረም ከፈለጉ እና እንደ ReplaceText ያሉ ተራ ካሬዎች ተስማሚ ካልሆኑ የራስዎን ስክሪፕት መፃፍ ይኖርብዎታል። ይህ የExecuteGroogyScript ካሬን በመጠቀም ሊከናወን ይችላል። የእሱ ቅንጅቶች ከዚህ በታች ይታያሉ.

ስክሪፕቱን ወደዚህ ካሬ ለመጫን ሁለት አማራጮች አሉ። የመጀመሪያው ስክሪፕት ያለው ፋይል በመስቀል ነው። ሁለተኛው ስክሪፕቱን ወደ ስክሪፕትቦዲ በማስገባት ነው። እኔ እስከማውቀው ድረስ፣ የፈጻሚው ስክሪፕት ካሬ ብዙ PLsን ይደግፋል - ከመካከላቸው አንዱ ግሩቭ ነው። የጃቫ ገንቢዎችን አሳዝኛለሁ - እንደዚህ ባሉ ካሬዎች ውስጥ ስክሪፕቶችን በጃቫ ውስጥ መጻፍ አይችሉም። በትክክል ለሚፈልጉት, የራስዎን ብጁ ካሬ መፍጠር እና ወደ NIFI ስርዓት መጣል ያስፈልግዎታል. ይህ አጠቃላይ ክዋኔ ከታምቡር ጋር ረጅም ጭፈራዎች የታጀበ ነው ፣ በዚህ ጽሑፍ ውስጥ አንነጋገርም ። እኔ የመረጠ ቋንቋ ነው። ከታች በሶፕ መልእክት ውስጥ መታወቂያውን በቀላሉ የሚያድስ የሙከራ ስክሪፕት አለ። ልብ ማለት ያስፈልጋል። ፋይልን ከflowFile ወስደህ አዘምነህ፣ እንደሚያስፈልግህ አትርሳ፣ አዘምነህ፣ እዚያ ላይ አስቀምጠው። ሁሉም ቤተ-መጻሕፍት አለመገናኘታቸውም ልብ ሊባል የሚገባው ጉዳይ ነው። አሁንም ከሊባዎቹ አንዱን ማስመጣት እንዳለቦት ሊሆን ይችላል። ጉዳቱ በዚህ ካሬ ውስጥ ያለው ስክሪፕት ለማረም በጣም ከባድ ነው። ከ NIFI JVM ጋር ለመገናኘት እና የማረም ሂደቱን ለመጀመር መንገድ አለ. በግሌ የሀገር ውስጥ መተግበሪያን አሄድኩ እና ከክፍለ-ጊዜው ፋይል ማግኘት አስመስያለሁ። ማረም በአገር ውስጥም ተከናውኗል። ስክሪፕቱን ሲጫኑ የሚመጡ ስህተቶች ጉግል ለማድረግ በጣም ቀላል ናቸው እና በ 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)
}

በእውነቱ ፣ የካሬው ማበጀት የሚያበቃው እዚህ ላይ ነው። በመቀጠል, የተሻሻለው ፋይል ፋይሉን ወደ አገልጋዩ ወደሚልከው ካሬው ይተላለፋል. ከዚህ በታች የዚህ ካሬ ቅንጅቶች አሉ።

የሶፕ መልእክት የሚተላለፍበትን ዘዴ እንገልፃለን. የት ነው የምንጽፈው። በመቀጠል, ይህ ሳሙና መሆኑን መግለፅ ያስፈልግዎታል.

እንደ አስተናጋጅ እና ድርጊት (soapAction) ያሉ አንዳንድ ንብረቶችን እንጨምራለን. አስቀምጥ፣ አረጋግጥ። የሶፕ ጥያቄዎችን እንዴት እንደሚልኩ ለበለጠ መረጃ፣ ይመልከቱ እዚህ

የ NIFI ሂደቶችን ለመጠቀም ብዙ አማራጮችን ተመልክተናል። እንዴት እንደሚገናኙ እና የእነሱ እውነተኛ ጥቅሞች ምንድ ናቸው. የተገመቱ ምሳሌዎች ፈታኝ ናቸው እና በውጊያ ውስጥ ከእውነተኛው ትንሽ ይለያያሉ። ይህ ጽሑፍ ለገንቢዎች የተወሰነ ጥቅም እንደሚኖረው ተስፋ አደርጋለሁ። ለሰጠህው አትኩሮት እናመሰግናለን. ማንኛውም ጥያቄ ካለዎት - ይጻፉ. መልስ ለመስጠት እሞክራለሁ።

ምንጭ: hab.com