sabeel
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
0
Fork 0
Code Pull Requests Activity

reimplemented of conditions cache as a non static variable

This commit is contained in:
Michele Artini 2021-10-05 09:20:37 +02:00
parent 0a9ef34b56
commit 8bbaa17335
2 changed files with 51 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/IndexNotificationsJob.java
View File

@ -44,14 +44,13 @@ public class IndexNotificationsJob {
private static final Logger log = LoggerFactory.getLogger(IndexNotificationsJob.class); private static final Logger log = LoggerFactory.getLogger(IndexNotificationsJob.class);
private static Map<String, Map<String, List<ConditionParams>>> conditionsForSubscriptions = new HashMap<>();
public static void main(final String[] args) throws Exception { public static void main(final String[] args) throws Exception {
final ArgumentApplicationParser parser = new ArgumentApplicationParser( final ArgumentApplicationParser parser = new ArgumentApplicationParser(
IOUtils IOUtils
.toString(IndexNotificationsJob.class .toString(
.getResourceAsStream("/eu/dnetlib/dhp/broker/oa/index_notifications.json"))); IndexNotificationsJob.class
.getResourceAsStream("/eu/dnetlib/dhp/broker/oa/index_notifications.json")));
parser.parseArgument(args); parser.parseArgument(args);
final SparkConf conf = new SparkConf(); final SparkConf conf = new SparkConf();
@ -88,16 +87,19 @@ public class IndexNotificationsJob {
final List<Subscription> subscriptions = listSubscriptions(brokerApiBaseUrl); final List<Subscription> subscriptions = listSubscriptions(brokerApiBaseUrl);
initConditionsForSubscriptions(subscriptions);
log.info("Number of subscriptions: " + subscriptions.size()); log.info("Number of subscriptions: " + subscriptions.size());
if (subscriptions.size() > 0) { if (subscriptions.size() > 0) {
final Map<String, Map<String, List<ConditionParams>>> conditionsMap = prepareConditionsMap(subscriptions);
final Encoder<NotificationGroup> ngEncoder = Encoders.bean(NotificationGroup.class); final Encoder<NotificationGroup> ngEncoder = Encoders.bean(NotificationGroup.class);
final Encoder<Notification> nEncoder = Encoders.bean(Notification.class); final Encoder<Notification> nEncoder = Encoders.bean(Notification.class);
final Dataset<Notification> notifications = ClusterUtils final Dataset<Notification> notifications = ClusterUtils
.readPath(spark, eventsPath, Event.class) .readPath(spark, eventsPath, Event.class)
.map((MapFunction<Event, NotificationGroup>) e -> generateNotifications(e, subscriptions, startTime), ngEncoder) .map(
(MapFunction<Event, NotificationGroup>) e -> generateNotifications(
e, subscriptions, conditionsMap, startTime),
ngEncoder)
.flatMap((FlatMapFunction<NotificationGroup, Notification>) g -> g.getData().iterator(), nEncoder); .flatMap((FlatMapFunction<NotificationGroup, Notification>) g -> g.getData().iterator(), nEncoder);
notifications notifications
@ -107,27 +109,28 @@ public class IndexNotificationsJob {
} }
} }
protected static void initConditionsForSubscriptions(final List<Subscription> subscriptions) { protected static Map<String, Map<String, List<ConditionParams>>> prepareConditionsMap(
subscriptions.forEach(s -> conditionsForSubscriptions.put(s.getSubscriptionId(), s.conditionsAsMap())); final List<Subscription> subscriptions) {
final Map<String, Map<String, List<ConditionParams>>> map = new HashMap<>();
subscriptions.forEach(s -> map.put(s.getSubscriptionId(), s.conditionsAsMap()));
return map;
} }
protected static NotificationGroup generateNotifications(final Event e, protected static NotificationGroup generateNotifications(final Event e,
final List<Subscription> subscriptions, final List<Subscription> subscriptions,
final Map<String, Map<String, List<ConditionParams>>> conditionsMap,
final long date) { final long date) {
final List<Notification> list = subscriptions final List<Notification> list = subscriptions
.stream() .stream()
.filter(s -> StringUtils.isBlank(s.getTopic()) || s.getTopic().equals("*") || s.getTopic().equals(e.getTopic())) .filter(
.filter(s -> verifyConditions(e.getMap(), conditionsAsMap(s))) s -> StringUtils.isBlank(s.getTopic()) || s.getTopic().equals("*") || s.getTopic().equals(e.getTopic()))
.filter(s -> verifyConditions(e.getMap(), conditionsMap.get(s.getSubscriptionId())))
.map(s -> generateNotification(s, e, date)) .map(s -> generateNotification(s, e, date))
.collect(Collectors.toList()); .collect(Collectors.toList());
return new NotificationGroup(list); return new NotificationGroup(list);
} }
private static Map<String, List<ConditionParams>> conditionsAsMap(final Subscription s) {
return conditionsForSubscriptions.get(s.getSubscriptionId());
}
private static Notification generateNotification(final Subscription s, final Event e, final long date) { private static Notification generateNotification(final Subscription s, final Event e, final long date) {
final Notification n = new Notification(); final Notification n = new Notification();
n.setNotificationId("ntf-" + DigestUtils.md5Hex(s.getSubscriptionId() + "@@@" + e.getEventId())); n.setNotificationId("ntf-" + DigestUtils.md5Hex(s.getSubscriptionId() + "@@@" + e.getEventId()));
@ -151,15 +154,18 @@ public class IndexNotificationsJob {
if (conditions.containsKey("trust") if (conditions.containsKey("trust")
&& !SubscriptionUtils && !SubscriptionUtils
.verifyFloatRange(map.getTrust(), conditions.get("trust").get(0).getValue(), conditions.get("trust").get(0).getOtherValue())) { .verifyFloatRange(
map.getTrust(), conditions.get("trust").get(0).getValue(),
conditions.get("trust").get(0).getOtherValue())) {
return false; return false;
} }
if (conditions.containsKey("targetDateofacceptance") && !conditions if (conditions.containsKey("targetDateofacceptance") && !conditions
.get("targetDateofacceptance") .get("targetDateofacceptance")
.stream() .stream()
.anyMatch(c -> SubscriptionUtils .anyMatch(
.verifyDateRange(map.getTargetDateofacceptance(), c.getValue(), c.getOtherValue()))) { c -> SubscriptionUtils
.verifyDateRange(map.getTargetDateofacceptance(), c.getValue(), c.getOtherValue()))) {
return false; return false;
} }

41
dhp-workflows/dhp-broker-events/src/test/java/eu/dnetlib/dhp/broker/oa/IndexNotificationsJobTest.java
View File

@ -5,10 +5,12 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.Arrays; import java.util.Arrays;
import java.util.List; import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import eu.dnetlib.dhp.broker.model.ConditionParams;
import eu.dnetlib.dhp.broker.model.Event; import eu.dnetlib.dhp.broker.model.Event;
import eu.dnetlib.dhp.broker.model.MappedFields; import eu.dnetlib.dhp.broker.model.MappedFields;
import eu.dnetlib.dhp.broker.model.Subscription; import eu.dnetlib.dhp.broker.model.Subscription;
@ -18,15 +20,19 @@ class IndexNotificationsJobTest {
private List<Subscription> subscriptions; private List<Subscription> subscriptions;
private Map<String, Map<String, List<ConditionParams>>> conditionsMap;
private static final int N_TIMES = 1_000_000; private static final int N_TIMES = 1_000_000;
@BeforeEach @BeforeEach
void setUp() throws Exception { void setUp() throws Exception {
final Subscription s = new Subscription(); final Subscription s = new Subscription();
s.setTopic("ENRICH/MISSING/PID"); s.setTopic("ENRICH/MISSING/PID");
s.setConditions("[{\"field\":\"targetDatasourceName\",\"fieldType\":\"STRING\",\"operator\":\"EXACT\",\"listParams\":[{\"value\":\"reposiTUm\"}]},{\"field\":\"trust\",\"fieldType\":\"FLOAT\",\"operator\":\"RANGE\",\"listParams\":[{\"value\":\"0\",\"otherValue\":\"1\"}]}]"); s
.setConditions(
"[{\"field\":\"targetDatasourceName\",\"fieldType\":\"STRING\",\"operator\":\"EXACT\",\"listParams\":[{\"value\":\"reposiTUm\"}]},{\"field\":\"trust\",\"fieldType\":\"FLOAT\",\"operator\":\"RANGE\",\"listParams\":[{\"value\":\"0\",\"otherValue\":\"1\"}]}]");
subscriptions = Arrays.asList(s); subscriptions = Arrays.asList(s);
IndexNotificationsJob.initConditionsForSubscriptions(subscriptions); conditionsMap = IndexNotificationsJob.prepareConditionsMap(subscriptions);
} }
@Test @Test
@ -34,7 +40,8 @@ class IndexNotificationsJobTest {
final Event event = new Event(); final Event event = new Event();
event.setTopic("ENRICH/MISSING/PROJECT"); event.setTopic("ENRICH/MISSING/PROJECT");
final NotificationGroup res = IndexNotificationsJob.generateNotifications(event, subscriptions, 0); final NotificationGroup res = IndexNotificationsJob
.generateNotifications(event, subscriptions, conditionsMap, 0);
assertEquals(0, res.getData().size()); assertEquals(0, res.getData().size());
} }
@ -46,7 +53,8 @@ class IndexNotificationsJobTest {
event.getMap().setTargetDatasourceName("reposiTUm"); event.getMap().setTargetDatasourceName("reposiTUm");
event.getMap().setTrust(0.8f); event.getMap().setTrust(0.8f);
final NotificationGroup res = IndexNotificationsJob.generateNotifications(event, subscriptions, 0); final NotificationGroup res = IndexNotificationsJob
.generateNotifications(event, subscriptions, conditionsMap, 0);
assertEquals(1, res.getData().size()); assertEquals(1, res.getData().size());
} }
@ -58,7 +66,8 @@ class IndexNotificationsJobTest {
event.getMap().setTargetDatasourceName("Puma"); event.getMap().setTargetDatasourceName("Puma");
event.getMap().setTrust(0.8f); event.getMap().setTrust(0.8f);
final NotificationGroup res = IndexNotificationsJob.generateNotifications(event, subscriptions, 0); final NotificationGroup res = IndexNotificationsJob
.generateNotifications(event, subscriptions, conditionsMap, 0);
assertEquals(0, res.getData().size()); assertEquals(0, res.getData().size());
} }
@ -68,14 +77,15 @@ class IndexNotificationsJobTest {
event.setTopic("ENRICH/MISSING/PROJECT"); event.setTopic("ENRICH/MISSING/PROJECT");
// warm up // warm up
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
final long start = System.currentTimeMillis(); final long start = System.currentTimeMillis();
for (int i = 0; i < N_TIMES; i++) { for (int i = 0; i < N_TIMES; i++) {
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
} }
final long end = System.currentTimeMillis(); final long end = System.currentTimeMillis();
System.out.println(String.format("no topic - repeated %s times - execution time: %s ms ", N_TIMES, end - start)); System.out
.println(String.format("no topic - repeated %s times - execution time: %s ms ", N_TIMES, end - start));
} }
@ -88,14 +98,15 @@ class IndexNotificationsJobTest {
event.getMap().setTrust(0.8f); event.getMap().setTrust(0.8f);
// warm up // warm up
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
final long start = System.currentTimeMillis(); final long start = System.currentTimeMillis();
for (int i = 0; i < N_TIMES; i++) { for (int i = 0; i < N_TIMES; i++) {
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
} }
final long end = System.currentTimeMillis(); final long end = System.currentTimeMillis();
System.out.println(String.format("topic match - repeated %s times - execution time: %s ms ", N_TIMES, end - start)); System.out
.println(String.format("topic match - repeated %s times - execution time: %s ms ", N_TIMES, end - start));
} }
@Test @Test
@ -107,14 +118,16 @@ class IndexNotificationsJobTest {
event.getMap().setTrust(0.8f); event.getMap().setTrust(0.8f);
// warm up // warm up
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
final long start = System.currentTimeMillis(); final long start = System.currentTimeMillis();
for (int i = 0; i < N_TIMES; i++) { for (int i = 0; i < N_TIMES; i++) {
IndexNotificationsJob.generateNotifications(event, subscriptions, 0); IndexNotificationsJob.generateNotifications(event, subscriptions, conditionsMap, 0);
} }
final long end = System.currentTimeMillis(); final long end = System.currentTimeMillis();
System.out.println(String.format("topic no match - repeated %s times - execution time: %s ms ", N_TIMES, end - start)); System.out
.println(
String.format("topic no match - repeated %s times - execution time: %s ms ", N_TIMES, end - start));
} }
} }