Join two datasets with a function similar to Vlookup [Big Dataset and Efficiency issue] - Wizards are welcome!

Question

I am working on a huge dataset and I have a lot of data formatted in XML, we are talking about a couple of GBs. For this reason I am currently using the stream function to read them and extract the information I need. My goal is to join two datasets with a Mathematica function similar to Vlookup. The first record from the stream looks like this

<row Id="1" PostId="1" Answer="2" CreationDate="2015-10-29T15:56:52.933" Score="13" ViewCount="200" UserId="23" LastEditorUserId="729" LastEditDate="2016-09-27T17:44:28.253" LastActivityDate="2016-09-27T17:44:28.253" Skills="&lt;.net&gt;&lt;asp.net-web-api&gt;&lt;asp.net&gt;" AnswerCount="1" CommentCount="2" FavoriteCount="1" /> 

And the second like this

<row Id="2" Level="3345" CreationDate="2008-07-31T14:22:31.287" LastAccessDate="2018-03-10T01:20:14.367" Location="Corvallis, OR"  Views="23041" Votes="646" Age="41" Sex="1" AccountId="2" />

Now what I would do is to create a list with the following elements: {"Id", "PostId", "CreationDate", "Score", "ViewCount", "UserId", "Skills", "Location"} and join the records where "UserId" and "AccountId" are equal. (i.e. one record has UserId=2 should be matched with the other which has AccountId=2)

My approach to this problem was:

1. First screen in the records
2. Save the minimum what I need
3. Do the same for the other record
4. Work on the Vlookup

However, my problems are relatively 2:

• I am new to Mathematica and I am not able to do a Vlookup
• I have something like 10 millions records, so efficiency is the priority

The code I used so far are the following.

SetStreamPosition[str, 0]; 
 list = Flatten[
   StringCases[{___ ~~ "Id=\"" ~~ ___, "CreationDate=\"" ~~ ___, 
      "Score=\"" ~~ ___, "ViewCount=\"" ~~ ___, "Skills=\"" ~~ ___}] /@ 
    StringSplit[
     Rest[ReadList[str, Record, RecordLists -> False, 
       WordSeparators -> {"\t", " "}]]] , 2]; 
 pos = Flatten[Position[StringTake[#, 1 ;; 3] & /@ list, "Id="]];
 AppendTo[pos, Length[list] + 1];
 data = MapAt[DateObject, data, {All, "CreationDate"}];
 data = Map[Association[Rule @@@ StringSplit[#, "="]] &, 
    Internal`PartitionRagged[list, Differences[pos]]];
 newdata = 
   data[[All, {"Id", "PostId", "CreationDate", "Score", 
      "ViewCount", "UserId", "Skills"}]];
 newdata = 
  MapAt[If[! MissingQ[#], 
     StringReplace[#, {"&lt;" -> "<", "&gt;" -> ">"}], #] &, 
   newdata, {All, "Skills"}]
 ]

and on the same style:

SetStreamPosition[str, 0]; 
     list = Flatten[
       StringCases[{___ ~~ "Id=\"" ~~ ___, "Location=\"" ~~ ___}] /@ 
        StringSplit[
         Rest[ReadList[str, Record, RecordLists -> False, 
           WordSeparators -> {"\t", " "}]]] , 2]; 
     pos = Flatten[Position[StringTake[#, 1 ;; 3] & /@ list, "Id="]];
     AppendTo[pos, Length[list] + 1];
     data = Map[Association[Rule @@@ StringSplit[#, "="]] &, 
        Internal`PartitionRagged[list, Differences[pos]]];
     newdata = 
       data[[All, {"AccountId", "Location"}]]

The last code is not perfect since it doesn't actually give the Location because it separated by comma it gives some problems.

My request for the Mathematica Stack Exchange world would be to help me figure out where actually the lines that are slowing down (and also why) and have a suggestion on how people more expert then me would achieve the goal of joining two datasets so big and with the following elements: {"Id", "PostId", "CreationDate", "Score", "ViewCount", "UserId", "Skills", "Location"}. Thanks in advance for your help and your time!

Answer 1

I am dealing with similar kinds of data, although in another format. Efficient code and parallelization really is key to acceptable performance. I will try to share my parallelized generic low-level importer when it is done. For the time being, some suggestions:

Importing

Do ReadList in chunks of lines and process those chunks. Use Sow and Reap for building the main list of data.

Reap[
   While[
     (lines = ReadList[stream, "String", 1000, NullRecords -> True]) =!= {}),
       (* line splitting and processing goes here *)
       StringCases[{___ ~~ "Id=\"" ~~ ___, ...},

       (* Sow the result for each line *)
       Sow[...]

Chunk processing makes parallelization easier to implement later. While it is certainly possible, parallelization is not as straight-forward as it might seem.

Sow and Reap are generally very efficient to construct lists.

A note to your code: When using WordSeparators -> {"\t", " "}, it is usually required to set NullWords -> True as well. (It is very unintuitive that NullWords is set to False by default as it leads to unexpected behaviour.)

Joining two datasets (the VLOOKUP)

Use Dispatch. It is very fast while everything else is very slow. A dispatch table is a list of replacement rules optimized for fast operation. When I need to "left join" table1 and table2 on columns col1 and col2, respectively, I use these little helper functions to create a dispatch table and join them:

Options[MakeDispatchTable] = {"AppendRules" -> {_ -> Missing[]}};
MakeDispatchTable[list_, column_, opts : OptionsPattern[]] := 
 Dispatch@MakeRuleTable[list, column, opts]

Options[MakeRuleTable] = Options[MakeDispatchTable];
MakeRuleTable[list_, column_, opts : OptionsPattern[]] := 
  Join[
    Rule[#[[Sequence @@ column]], #] & /@ list, 
    OptionValue@"AppendRules"
  ];

table12joined = MapThread[Append,
  {table1, Replace[table1[[All, col1]], MakeDispatchTable[table2, col2], {1}]}
]

Note that the Rule _ -> Missing[] is appended to replace any row which did not match with a Missing[]. This is also why it is important to use Replace[..., ..., {1}], so it will only replace on level one, instead of using /., matching the whole expression with _ and replacing it with Missing[] altogether.

EDIT

As of version 10.0, there is the built-in function Merge which provides the functionality described with the dispatch table above and features good performance.

Answer 2

As only one line from each file was provided as a record, a string template was created to generate multiple rows/lines.

 Clear[table1template, table1rows, table1, table1ds, table2template,table2rows, table2, table2ds, tablejoin, tablemerge];
table1template = StringTemplate["<row Id=\"`id`\" PostId=\"`postid`\" Answer=\"`answer`\" CreationDate=\"`creationdate`\" Score=\"`score`\" ViewCount=\"`viewcount`\" UserId=\"`userid`\" LastEditorUserId=\"`lasteditoruserid`\" LastEditDate=\"`lasteditdate`\" LastActivityDate=\"`lastactivitydate`\" Skills=\"`skills`\" AnswerCount=\"`answercount`\" CommentCount=\"`commentcount`\" FavoriteCount=\"`favoritecount`\" />\n"];
table1rows = StringJoin[Table[TemplateApply[table1template,Association[
    Rule["id", RandomInteger[{1,100}]], 
    Rule["postid", RandomInteger[{1,10}]],
    Rule["answer", RandomInteger[{1,5}]],
    Rule["creationdate",StringJoin[DateString[DatePlus[DateList[],{-RandomInteger[{1,100}],"Month"}], "ISODateTime"], ".", ToString[RandomInteger[{0,999}]]]],
    Rule["score",RandomInteger[{1,50}]],
    Rule["viewcount",RandomInteger[{1,400}]],
    Rule["userid",i (*RandomInteger[{1,20}]*)],
    Rule["lasteditoruserid",RandomInteger[{500,999}]],
    Rule["lasteditdate",StringJoin[DateString[DatePlus[DateList[],{-RandomInteger[{1,100}],"Month"}], "ISODateTime"], ".", ToString[RandomInteger[{0,999}]]]],
    Rule["lastactivitydate",StringJoin[DateString[DatePlus[DateList[],{-RandomInteger[{1,10}],"Month"}], "ISODateTime"], ".", ToString[RandomInteger[{0,999}]]]],
    Rule["skills",StringJoin[RandomChoice[{"&lt;.net&gt;","&lt;asp.net-web-api&gt;", "&lt;asp.net&gt;"}, {1,5}]]],
    Rule["answercount",RandomInteger[{1,5}]],
    Rule["commentcount",RandomInteger[{1,5}]],
    Rule["favoritecount",RandomInteger[{1,5}]]
]
], {i,10}]];

table2template = StringTemplate["<row Id=\"`id`\" Level=\"`level`\" CreationDate=\"`creationdate`\" LastAccessDate=\"`lastaccessdate`\" Location=\"`location`\"  Views=\"`views`\" Votes=\"`votes`\" Age=\"`age`\" Sex=\"`sex`\" AccountId=\"`accountid`\" />\n"];
table2rows = StringJoin[Table[TemplateApply[table2template,Association[
    Rule["id", RandomInteger[{1,100}]], 
    Rule["level", RandomInteger[{1,10}]],
    Rule["creationdate",StringJoin[DateString[DatePlus[DateList[],{-RandomInteger[{1,100}],"Month"}], "ISODateTime"], ".", ToString[RandomInteger[{0,999}]]]],
    Rule["lastaccessdate",StringJoin[DateString[DatePlus[DateList[],{-RandomInteger[{1,100}],"Month"}], "ISODateTime"], ".", ToString[RandomInteger[{0,999}]]]],
    Rule["location", RandomChoice[{"Portland​, OR","Eugene​​, OR","Beaverton​​, OR","Bend​​, OR","Corvallis​​, OR","Medford​​, OR","Gresham​​, OR","Lake Oswego​​, OR","Ashland​​, OR","Hillsboro​​, OR","Salem​​, OR","Sherwood​​, OR","Oregon City​​, OR","Grants Pass​​, OR","West Linn​​, OR","Tigard​​, OR","McMinnville​​, OR"}, 1]],
    Rule["views",RandomInteger[{1,50}]],
    Rule["votes",RandomInteger[{1,400}]],
    Rule["age",RandomInteger[{18,65}]],
    Rule["sex",RandomInteger[{1,2}]],
    Rule["accountid",i (*RandomInteger[{1,20}]*)]
]
], {i,10}]];

We generate 10 lines of each file types. So if you were to chunk this to say 1000 records/lines I would suggest output each chunk to file with CreateFile[] then OpenAppend[]. But we can try a run with mock data to see how we could either Join the datasets or Merge the datasets.

    table1 = ReplaceAll[
 Cases[ImportString[StringJoin["<root>", table1rows, "</root>"], 
   "XML"], XMLElement["row", {atts__}, {}] :>  {atts}, Infinity], 
 KeyValuePattern[{Rule["Id", id_] ,
    Rule["PostId", postid_],
    Rule["CreationDate", creationdate_] , 
    Rule["Score", score_] ,
    Rule["ViewCount", viewcount_] ,
    Rule["UserId", userid_] ,
    Rule["Skills", skills_]}] :> {id, postid, DateObject[creationdate], score, viewcount, userid, skills}];

table2 = ReplaceAll[Cases[ImportString[StringJoin["<root>",table2rows, "</root>"], "XML"], XMLElement["row",{atts__},{}]:>  {atts}, Infinity], KeyValuePattern[{Rule["AccountId", accountid_] ,Rule["Location", location_]}] :> {accountid, location}];

table1ds = Dataset[table1];
table2ds = Dataset[table2];

To use Join

    tablejoin = Transpose[
  Drop[Transpose[
    Map[Flatten[#] &, 
     Transpose[
      Drop[Transpose[
        ReplaceAll[
         Normal[Map[Flatten[#] &, 
           Join[Normal[table1ds[[1 ;; -1]][GroupBy[#[[6]] &]]], 
            Normal[table2ds[[1 ;; -1]][GroupBy[#[[1]] &]]], 2]
           ]
          ], Rule -> List]], 1]]]], {8}]];

To use Merge

    tablemerge = Transpose[
  Drop[Transpose[
    Map[Flatten[#] &, 
     Transpose[
      Drop[Transpose[
        ReplaceAll[
         Normal[Map[Flatten[#] &, 
           Merge[{Normal[table1ds[[1 ;; -1]][GroupBy[#[[6]] &]]], 
             Normal[table2ds[[1 ;; -1]][GroupBy[#[[1]] &]]]}, 
            Identity]]]
         , Rule -> List]], 1]]]], {8}]];

So I would suggest reading in the files in chunks and appending each filtered chunks to file as CSV. This should results in much smaller sized files. The files can then be imported whole. As for cases when there is no AccountId in file2 matching UserId in file1 is not handled in this example.