D-Net

<?xml version="1.0"?>

<workflow-app xmlns="uri:oozie:workflow:0.4" name="websiteusage_document_communities_builder">

	<parameters>

		<property>

			<name>input_logs</name>

			<description>input logs file</description>

		</property>

		<property>

			<name>output</name>

			<description>output publication to communities mappings</description>

		</property>

	</parameters>

	<start to="communities_builder" />

    <action name="communities_builder">

        <map-reduce>

            <job-tracker>${jobTracker}</job-tracker>

            <name-node>${nameNode}</name-node>

            <prepare>

                <delete path="${nameNode}${output}"/>

            </prepare>

            <streaming>

                <mapper>scripts/madis/mexec.py -f scripts/createCommunities.sql -w communities.db</mapper>

            </streaming>

            <configuration>

                <property>

                    <name>mapred.output.format.class</name>

                    <value>com.cloudera.science.avro.streaming.AvroAsJSONOutputFormat</value>

                </property>

                <property>

                    <name>mapred.input.format.class</name>

                    <value>com.cloudera.science.avro.streaming.AvroAsJSONInputFormat</value>

                </property>

                <property>

                    <name>mapred.reduce.tasks</name>

                    <value>0</value>

                </property>

                <property>

                    <name>mapred.input.dir</name>

                    <value>${input_logs}</value>

                </property>

                <property>

                    <name>eu.dnetlib.iis.avro.input.class</name>

                    <value>eu.dnetlib.iis.websiteusage.schemas.LogEntry</value>

                </property>

                <property>

                    <name>mapred.output.dir</name>

                    <value>${output}</value>

                </property>          

                <property>

                    <name>eu.dnetlib.iis.avro.output.class</name>

                    <value>eu.dnetlib.iis.websiteusage.schemas.DocumentToCommunity</value>

                </property>

				<property>

					<name>mapred.task.timeout</name>

					<value>1800000</value>

				</property>

				<property>

					<name>mapreduce.task.timeout</name>

					<value>1800000</value>

				</property>

            </configuration>

            <!-- 

            <file>${input_dataset_db}#dataset.db</file>

             -->

        </map-reduce>

        <ok to="end"/>

        <error to="fail"/>

    </action>

    <kill name="fail">

        <message>Unfortunately, the process failed -- error message:

        			[${wf:errorMessage(wf:lastErrorNode())}]

        		</message>

    </kill>

    <end name="end"/>

</workflow-app>

## This is a classpath-based import file (this header is required)

madis classpath eu/dnetlib/iis/3rdparty/scripts/madis

drop table if exists piwiklog;

create table piwiklog (datetime, action,user,session,data);

insert into piwiklog select * from (setschema 'datetime, action,user,session,data' select jdictsplit(stripchars(c1,','),'timestamp','action','user','session','data') from stdinput());

create index piwiklog_index on piwiklog(action,data,session,user,datetime);

select jdict('cid',cliqueid,'SimilarDocid',nodeid) from 

    (select graphcliques(node1,node2) from 

        (select pgroup1 as node1, pgroup2 as node2, count(*) as frequency from 

            (setschema 'session,pgroup1,pgroup2' select session,jcombinations(jsort(jset(jgroup(data))),2) as pgroup from 

                (select datetime, action,user,session,jlengthiest(jdictvals(urlquery2jdict(data),'personId','projectId','publicationId','datasourceId','datasetId')) as data from piwiklog) 

                where (action='viewPerson' or action='viewPublication' or action='viewProject' or action='viewDatasource' or action='viewOrganization' or action='viewDataset') 

            group by session) 

        group by node1,node2 having frequency>4)

    ) 

where nodeid is not null;

input_logs=/share/import/logs/2014-11-03

output=${workingDir}/out