Using Lag and Lead in Dataiku




I have time-series data (essentially a history recording a series of three events).  I want to convert these three events into a single record using either or both a lead or lag type function and specific time conditions. Let me provide an example, where ID is the unique case number, EVENT# is which event was recorded, and DAYS is the number of days from an initial event (not recorded):

ID  EVENT#  DAYS

01  1              30

01   2              45

01   3              80

02  1               30

02   2              62

02   3              99

The condition I want to test for to identify a dummy variable goes something like this:

Within an ID, (if the difference in DAYS for Event = 2 compared to Event = 1 is between (1,30)) AND (if the difference in DAYS for Event = 3 compared to Event = 2 is > 30) then the dummy =1, else = 0

So in the two examples, above, ID = 01, Dummy =1 because both conditions are met (condition 1 = 15 and condition 2 = 35), but with ID = 02, Dummy =0 because the first condition was not met (condition 1 = 32, condition 2 = 37).

I can do this in SAS by working from Event = 3 and lagging once for condition 2 and using two separate lag functions for condition 1 or by separating the conditions, using the lag function, then aggregating the results, but I need to do this within Dataiku.
Hi JimCreech,


It is possible to achieve a similar result in Dataiku by going through the following steps:


* run a [Window](https://knowledge.dataiku.com/latest/courses/visual-recipes/window.html) recipe using id as partitioning column, event as ordering column, and compute the lagdiff values on days
* run a [Prepare](https://knowledge.dataiku.com/latest/courses/basics/prepare-data/concept-prepare-recipe.html) recipe which will:
	+ run the assertions on the specific conditions you mentioned and create a dummy\_tmp column we'll reuse at the final step
	+ filter out the records where event == 01 (since we won't need it for the aggregation)
* run a [Groupby](https://knowledge.dataiku.com/latest/courses/basics/group-data/group-summary.html) recipe to aggregate at the id level and get the minimum of the dummy\_tmp column (which should be 1 if all conditions you described are verified, 0 otherwise).


You will find attached a sample DSS project with an example of implementation of those steps using the small example data mentioned in your post. If you are working on a regular basis with time series data, you may find this [tutorial](https://knowledge.dataiku.com/latest/courses/visual-recipes/window/window.html) on the Window recipe capabilities useful.


Hope this helps !


Best,


Harizo




[View solution in original post](/t5/General-Discussion/Using-Lag-and-Lead-in-Dataiku/m-p/19308#M1146)
java.sql.timestamp is a subclass of java.util.date - it should work the same for comparison purposes:

<https://docs.oracle.com/en/java/javase/11/docs/api/java.sql/java/sql/timestamp.html>

if you specifically need a date, you could convert it back pretty easily; check if isinstance(java.sql.timestamp… Hi,  

  

check the following Code  

SQLCopy Code

```
-- Question 1 Answer
SELECT E.EmployeeID,E.Day, (E.TotalTime/60) 'TotalHours', (E.TotalTime%60) 'Minites'
FROM (SELECT EmployeeID,CONVERT(VARCHAR(10),FromTime,102) 'Day',SUM(DATEDIFF(mi,FromTime,ToTime)) 'TotalTime'
      FROM tblEmployeeTimeCard
      GROUP BY EmployeeID,CONVERT(VARCHAR(10),FromTime,102))E

-- Question 1&2 Answer
SELECT E.EmployeeID,E.Day, (E.TotalTime/60) 'TotalHours', (E.TotalTime%60) 'Minites',
CASE WHEN (E.TotalTime/60)>=6 THEN E.TotalTime - (30* ROUND((E.TotalTime/60*6),1))
     ELSE (E.TotalTime/60) END 'TotalWorkHours'
FROM (SELECT EmployeeID,CONVERT(VARCHAR(10),FromTime,102) 'Day',SUM(DATEDIFF(mi,FromTime,ToTime)) 'TotalTime'
      FROM tblEmployeeTimeCard
      GROUP BY EmployeeID,CONVERT(VARCHAR(10),FromTime,102))E
```
  

  

For ur Question 2, I am not getting. give some more details about that. if possible send some sample data. if access over lap like (From date is 2013 -04-05 11:55:00PM to 2013-04-05 19:10:00PM)  

then we can't able to fine the time difference. So check the code for Question No 1 and give the feedback.  

  

Regards,  

GVPrabu A couple of notes to get started:



> 
> .apply(literal\_eval)
> 
> 
> 


Wouldn't this be better as `map`?



> 
> I then re-create the Dask DF:
> 
> 
> ddf = dd.from\_pandas(ddf\_out, npartitions=nCores)
> 
> 
> 


`ddf_out` was already a dask dataframe, I don't know why you had to do this.



> 
> The columns in result seem to match those in metadir.
> 
> 
> 


The value of `result.columns` is taken from the meta you provided, no computing happens until you ask for it (dask is lazy in most operations). Does the ValueError exception not provide further information?


Here is a full example



```
x = ({'F9_07_PZ_COMP_DIRECT': '0',
  'F9_07_PZ_DIRTRSTKEY_NAME': 'DEBRA MEALY',
  'F9_07_PZ_COMP_OTHER': '0',
  'F9_07_PZ_COMP_RELATED': '0',
  'F9_07_PZ_TITLE': 'CHAIR PERSON',
  'F9_07_PZ_AVE_HOURS_WEEK': '1.00',
  'F9_07_PC_TRUSTEE_INDIVIDUAL': 'X'},
 {'F9_07_PZ_COMP_DIRECT': '0',
  'F9_07_PZ_DIRTRSTKEY_NAME': 'HELEN GORDON',
  'F9_07_PZ_COMP_OTHER': '0',
  'F9_07_PZ_COMP_RELATED': '0',
  'F9_07_PZ_TITLE': 'VICE CHAIR',
  'F9_07_PZ_AVE_HOURS_WEEK': '1.00',
  'F9_07_PC_TRUSTEE_INDIVIDUAL': 'X'})
df = pd.DataFrame({'a': x})
d = dd.from_pandas(df, 1)
meta = pd.DataFrame(columns=['F9_07_PZ_COMP_DIRECT', 
       'F9_07_PZ_DIRTRSTKEY_NAME',
       'F9_07_PZ_COMP_OTHER', 'F9_07_PZ_COMP_RELATED', 'F9_07_PZ_TITLE',
       'F9_07_PZ_AVE_HOURS_WEEK', 'F9_07_PC_TRUSTEE_INDIVIDUAL'], dtype="O")
d.map_partitions(lambda df: df.a.apply(pd.Series), meta=meta).compute()

```

How did I know what `meta` to use? I applied the function to the pandas dataframe - you might use a small piece of the dataframe to do this.


Some additional notes:


* it is an anti-pattern to load the data with pandas, pass to dask workers and then gather the whole result back to a pandas (in memory) dataframe, you are very unlikely to see a speedup this way and may incur a lot of overhead. You are better off loading with something like `dd.read_csv`, and aggregating or writing with dask functions too. Only `compute()` on something that will be small or not return anything (because it involves writing the output). The official examples do not use from\_pandas.
* string and dict processing are python methods, and so hold the interpreter lock (GIL) of any python function: threads will not actually run in parallel. To get parallelism, you need to run in processes, which is most easily achieved using <https://docs.dask.org/en/latest/setup/single-distributed.html>
* the distributed scheduler also gives you access to the dashboard, which has a lot of useful information to diagnose how your system is running. You can also configure a lot about its behaviour, in case you have firewall rules you need to obey. let table = base.gettable("table 1")

let record = await input.recordasync("record", table)

await table.updaterecordasync(record.id, {

"single select": null

}) Reason: `s` starts at 1 and is incremented on first call, but is never reset. On seconds and subsequent calls `s` no longer starts at 1. `s` is actually unnecessary as you can use the loop counter `i` instead.



A more conventional way to do this sort of thing would be with a single function, not embedded in an object, and to control the detailed behaviour by passing parameter(s).



```
function loadGame(level){
	var values;
	switch(level){
		case 'hard':
			values: [7,8,9],
		break;
		case 'normal':
			values: [4,5,6],
		break;
		default://easy
			values: [1,2,3],
	}
	for(i=0;i<=this.values.length;i++){
		document.getElementById('n'+i).innerHTML=this.values[i];
	}
}
```


```
<div id="n0"></div>
	<div id="n1"></div>
	<div id="n2"></div>
	<input type="button" value="easy" onClick="loadGame('easy')"/>
	<input type="button" value="normal" onClick="loadGame('normal')"/>
	<input type="button" value="hard" onClick="loadGame('hard')"/>
```

**Airshow** one idea can be videos have frames so, vectorize each frame and calculate dtw between two consecutive frames. this will help in gaining the difference in each frame and perhaps help in classification. hope this helps you. Looks like YYMMDD to me. I would try to verify this in the documentation or whatever information source you have on the old database. You could also look for inappropriate numbers like MM not between 1 and 12 and DD not between 1 and 31. If you find any inappropriate numbers than this is the wrong approach. This does not look like a date based on an offset number of days after a start date because the start date would have to be about 2600 years ago. Your test is redundant, since you are already querying only for matching records. In this case, you will get no rows if there is no match, therefore you will never hit the else clause in the loop.  

  

Instead, it is enough to check if there are any matching records at all. The example below shows how (note that you can put a condition straight into Any() instead of specifying Where()).  

  

C#Copy Code

```
using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using System.Web.UI;
using System.Web.UI.WebControls;
 
namespace CustomerMaster
{
    public partial class CustomerMaster : System.Web.UI.Page
    {
 
        private void AddData()
        {
            
            OperationsDataContext AddDb = new OperationsDataContext();
            CustomerMast cm = new CustomerMast();
            
                        
            cm.CustomerCode = txtCustCde.Text;
            cm.CustomerName = txtCustName.Text;
            cm.CustomerAddress = txtCustAdd.Text;
            cm.BrkrCode = txtBrkCode.Text;
            cm.BrokerName = txtBrkName.Text.
            cm.CrrCode = txtCrrCode.CarrierCode;
            cm.CarrierName =  txtCrrName.Text;
            cm.StnCode=sm.StationCode;
            cm.StationName = txtStnName.Text;
            cm.VatNo=txtVatNo.Text;
            cm.TinNo=txtTinNo.Text;
            cm.TMCO=txtTMCONo.Text;
            cm.PanNo = txtPAN.Text;
			
	    if(AddDb.CustomerMasts.Any(c=>c.CustomerCode == txtCustCde.Text))
            {
	        Response.Write("Error! CustomerCode Exits");
	    }
	    else
	    {
                AddDb.BrokerMasts.InsertOnSubmit(bm);
                AddDb.CarrierMasts.InsertOnSubmit(crm);
                AddDb.StationMasts.InsertOnSubmit(sm);
                AddDb.CustomerMasts.InsertOnSubmit(cm);

	        AddDb.SubmitChanges();
            }         
        }
 
        protected void btnAdd\_Click(object sender, EventArgs e)
        {
            AddData();
        }
    }
}
``` [@najoua.abbaci](/u/najoua.abbaci)

mabye the flow can be adopted to following:

assign:

leftside: drresult (datatype: datarow)

right side:

dt\_datepv.asenumerable.orderby(function ( r ) datetime.parseexact(r(0).tostring.trim, “dd.mm.yyyy”, system.globalization.cultureinfo.currentuiculture.datetimeformat)).last()

so… Turns out If I tweked each method 



```
foreach (var dataColumn in readFields.Select(column => new DataColumn(columntypeof(string)) { AllowDBNull = true, DefaultValue = string.Empty }))
{
    csvData.Columns.Add(dataColumn);
}

```

and in the second Method I changed it to use an index rather than column name



```
for (var count = 0; count < table.Columns.Count; count++)
{
    sqlBulkCopy.ColumnMappings.Add(count, count);
}

```