Microsoft Stream Insight User Defined operator (UDO) didn't work

Question

I am using User Defined operator (UDO) to insert new fields into my incoming stream.

But when i use UDO in my StreamInsight application i got this exception :

" The management service encountered an error: Value cannot be null. Parameter name: nullableType " .

I tried to solve , but i couldn't . i don't know what is the real problems ?

Can anyone help me to solve this issue ?

Thanks in advance.

Kajan.

Answer 1

Could you please post the UDO code and the query?

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 2

Thanks DevBiker, 
1. UDO Part
public class SparkVhrmsUDO : CepOperator<SqlNewJoinIn, SqlNewJoinOut>
        {
            public overrideIEnumerable<SqlNewJoinOut> GenerateOutput(IEnumerable<SqlNewJoinIn> payloads)
            {
                List<SqlNewJoinOut> outputCollection = newList<SqlNewJoinOut>();
                foreach (SqlNewJoinIn payload in payloads)
                {
                    SqlNewJoinOut output = newSqlNewJoinOut();
                    output.AllocationEndDate = payload.AllocationEndDate;
                    output.AllocationStartDate = payload.AllocationStartDate;
                    output.CEG = payload.CEG;
                    output.Current_ATC = payload.Current_ATC;
                    output.description = "testing1";
                    output.Designation = payload.Designation;
                    output.EmpId = payload.EmployeeId;
                    output.FirstName = payload.FirstName;
                    output.JoinDate = payload.JoinDate;
                    output.LastName = payload.LastName;
                    output.MaterialsUrl = "testing2";
                    output.OfficialEMail = payload.OfficialEMail;
                    output.Permanent_ATC = payload.Permanent_ATC;
                    output.Project_Manager = payload.Project_Manager;
                    output.ProjectName = payload.ProjectName;
                    output.ProjectRole = payload.ProjectRole;
                    output.Reporting_Manager = payload.Reporting_Manager;
                    output.Tier = payload.Tier;
 
                    outputCollection.Add(output);
                }
                return outputCollection;
            }
        }

        [CepUserDefinedOperator(typeof(Program.SparkVhrmsUDO))]
        publicstaticSqlNewJoinOut VhrmsUDO(thisCepWindow<SqlNewJoinIn> window)
        {
            throwCepUtility.DoNotCall();   
        }
2. Query Part
 var kpiQuery2 = from o in streamSource2.CountByStartTimeWindow(2,CountWindowOutputPolicy.PointAlignToWindowEnd)
                 select o.VhrmsUDO();
   

      

Answer 3

OK ... the UDO itself looks correct. Do you have a stack trace from the exception? And are any of the types on your payloads nullable? It looks like it may be an issue with the internal registration that StreamInsight does of the payload types but, at this point, I can't be certain. Perhaps a quick example of your types would also be helpful.

Now ... on the other hand, what is it that you are trying to do here? I know that this is a mock up but what is it that you intend for it to do when completed?

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 4

Thanks for your feedback DevBiker,

My task is get the stream from input source and add additional fields. value of that additional fields will be determine at run time according to the input stream parameters. then send it to output adapter.

 This is the stack trace i got when i get that exception,

at Microsoft.ComplexEventProcessing.Diagnostics.Exceptions.Throw(Exception exception)
   at Microsoft.ComplexEventProcessing.RemoteServerProxy.Create(XmlReader definition, Uri parentName)
   at Microsoft.ComplexEventProcessing.Server.CreateCepObject(CepObject obj)
   at Microsoft.ComplexEventProcessing.Application.CreateQueryTemplate(XmlReader queryTextReader, CepStreamSnapshot snapshot)
   at Microsoft.ComplexEventProcessing.Application.CreateQueryTemplate(String queryTemplateName, String description, CepStreamSnapshot snapshot)
   at Microsoft.ComplexEventProcessing.Application.CreateQuery[TPayload](String queryName, String queryDescription, CepStreamSnapshot snapshot, CepStreamConsumer`1 consumer, Boolean isResilient)
   at Microsoft.ComplexEventProcessing.Linq.CepStream`1.ToQuery(Application application, String queryName, String description, Type adapterFactoryType, Object adapterConfig, EventShape eventShape, StreamEventOrder eventOrder, Boolean isResilient)
   at Microsoft.ComplexEventProcessing.Linq.CepStream`1.ToQuery(Application application, String queryName, String description, Type adapterFactoryType, Object adapterConfig, EventShape eventShape, StreamEventOrder eventOrder)
   at SPARK_Standalone.Program.CreateAndStartQuery[T](CepStream`1 sourceStream, String queryName, String queryDescription, EventShape eventShape, SqlOutputConfig outputConfig, Application cepApplication, Server cepServer, Boolean start) in C:\Users\Kajan\Desktop\SPARK\stanaloneserver\server\SPARK_Standalone\SPARK_Standalone\Program.cs:line 466
   at SPARK_Standalone.Program.Start_SQL_Query(SqlInputConfig inputConfig, SqlOutputConfig outputConfig, AdvanceTimeSettings inputAdvaceTimeSettings, Int32 i) in C:\Users\Kajan\Desktop\SPARK\stanaloneserver\server\SPARK_Standalone\SPARK_Standalone\Program.cs:line 393
   at SPARK_Standalone.Program.StartAppAndQueries(Server cepServer) in C:\Users\Kajan\Desktop\SPARK\stanaloneserver\server\SPARK_Standalone\SPARK_Standalone\Program.cs:line 271
   at SPARK_Standalone.Program.Main(String[] args) in C:\Users\Kajan\Desktop\SPARK\stanaloneserver\server\SPARK_Standalone\SPARK_Standalone\Program.cs:line 35
   at System.AppDomain._nExecuteAssembly(RuntimeAssembly assembly, String[] args)
   at System.AppDomain.ExecuteAssembly(String assemblyFile, Evidence assemblySecurity, String[] args)
   at Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly()
   at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
   at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
   at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
   at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
   at System.Threading.ThreadHelper.ThreadStart()

But when i try the same application in embedded Server method (use server.create instead of server.connect) it works fine without any error. it means i don't have any problems in code.

so, Why do i get the exception when i try same application in standalone server method?

Answer 5

It's got to have to do with the serialization of the type across the processes ... and it looks like something with the query template.

Question ... do you have the DLLs for the UDO and your other code deployed on the remote server? If you are using the standalone server on the same machine, have you copied the DLLs to the bin folder for the standalone service? That would certainly explain why you get the exception with the standalone server but not with the embedded server.

Also ... this sounds like you are trying to match your data stream with reference data. This is better accomplished within the StreamInsight engine, not with a UDO. Check out this blog post.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 6

Now it s working DevBiker,

Reason was Stream insight server couldn't locate the assembly  at run time. As you said i copied the DLL files to Host folder of stream insight. after that i could locate the DLL files at run time. Thanks a lot for your help to solve this issue.

Another problem i faced was , i have 3 projects in my solution , two adapters (SQL and Oracle) and one main project (Start up project). but if i put those functions in main project it will not work, because i don't have DLL of main project to copy to Host folder. but if i put those functions in somewhere else (SQL or Oracle adapter project) it will work. Which locations is correct for those functions?

Now i am matching my data stream with reference data (No more UDO :) ). function wise it is working. but which is the best method ?

Thanks & regards,

Kajaruban.S

Answer 7

The best method is to use the StreamInsight engine and the reference data pattern. It's more efficient and causes less latency.

As for your projects ... even if it's an EXE you can copy that to the host folder. But I'm guessing that the "main project" that you mention is merely a test harness, correct? Or is that the project that calls to the remote StreamInsight server? Either way, actually, I would put the adapters in a separate assembly. Have an assembly for each is fine but I'd put them in the same assembly ... but that's really just a matter of personal preference and not something that needs to be done. The important thing, as you've discovered, is that the assembly needs to be available to the remote server instance. Any extensions ... UDO's, UDA's, UDSO's, etc. ... also need to be available on the remote server. I'd put those in one of your DLLs and have a clean separation between code that runs in the engine (adapters, extensions) and code that controls the engine (your main project). Your queries are, btw, serialized when connecting remotely so they don't have to be available on the remote instance.

I hope that helps.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 8

ok , thanks DevBiker,

Connect with remote server , create the CEP application , define the query logic and all other parts are written in main project and i have a different assembly for each adapter.

Yeah , if i use any user defined function (For example, output.description = User.FindDescription(output.JoinDate); ) to match my data stream with reference data  , those DLLs needs to be available to the remote server instance. otherwise we dont need to copy Dlls to host folder . Am i correct?

You are correct , i will divide my code into two parts . that is the best method. Thank you for your Advise.

Thanks & Regards,

Kajaruban.S