Работа с архитектурой TEDA 4.2. Как вызвать оператор LookUpCache в цикле?

Question

Работа с архитектурой TEDA 4.2. Как вызвать оператор LookUpCache в цикле?

У меня есть вход как a1;b1;c1|a2;b2;c2|a3;b3;c3

Теперь я хочу использовать a1, a2 и a3 в качестве ключа поиска и генерировать кортеж для каждого (в качестве вывода). Я извлек a1,a2,a3, используя токенизатор, но как передать его оператору LookupCache, чтобы он использовал все ключи один за другим и генерировал 3 кортежа.

Answer 1

Используя платформу приложения TEDA, вам необходимо подготовить схему TypesCommon.ReaderOutStreamType, которая является выходом синтаксического анализа и вводом данных. Он расширен атрибутами, указанными в типе xxx.streams.custom::TypesCustom.LookupType.

В xxx.streams.custom.TypesCustom.spl

    static LookupType = tuple<
        // ------------------------------------------------
        // custom code begin
        // ------------------------------------------------
        // add your custom attributes here
        // key attribute for LookupCache, e.g. a1, a2, a3 from input
        rstring a
        // ------------------------------------------------
        // custom code end
        // ------------------------------------------------
    >;

В xxx.chainprocessor.transformer.custom.DataProcessor.spl вы можете подготовить извлекать ключевой атрибут для вашего поиска, заполнить a атрибут и отправить n кортежей (LookupStream) оператору LookupCache из одного входного кортежа (InRec)

    (stream<I> LookupStream as O) as PrepareLookup = Custom(InRec as I) {
        logic
            onTuple I: {
                mutable list<rstring> list1 = tokenize(attrInput, "|", true);
                for (rstring val in list1) {
                    mutable list<rstring> list2 = tokenize(val, ";", true);
                    I.a = list2[0];
                    submit(I, O); // submit for a1, a2 ...
                }   
            }
            onPunct I: {
                if (currentPunct() == Sys.WindowMarker) {
                    submit(Sys.WindowMarker, O);
                }
            }
    }


    stream<I> EnrichedRecords = LookupCache(LookupStream as I) {
        param
            segmentName:    "yourSeg"; // The name of the physical memory segment
            containerName:  "yourContainer"; // The name of the store
            useMutex:   false; // Don't protect data with mutex
            mutexName:  "";
            key:        I.a;
            keyType:        rstring;           // SPL type of key
            found:      I.lookupFound;     // attribute that takes lookup success
            valueType:  TypesCustom.YourType_t; //type of lookup table value as also defined in LookupManager
            disableLookup:  $disableLookup;   // always pass through this parameter
            applControlDir: $applControlDir;   // always pass through this parameter
    }