Elgato_dark/zeppelin
1
0
Fork 0
mirror of https://github.com/apache/zeppelin synced 2026-05-24 09:38:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Address review feedbacks

Browse source 
Signed-off-by: karuppayya <karuppayya1990@gmail.com>
This commit is contained in:
karuppayya 2016-12-01 21:34:00 +05:30 committed by Karup
parent 1a4528405c
commit d4e54e8cad
10 changed files with 100 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
View file

@ -166,8 +166,8 @@ public class SparkInterpreter extends Interpreter {
int jobId = jobStart.jobId();
String jobGroupId = jobStart.properties().getProperty("spark.jobGroup.id");
String jobUrl = getJobUrl(jobId);
String noteId = getNoteId(jobGroupId);
String paragraphId = getParagraphId(jobGroupId);
String noteId = Utils.getNoteId(jobGroupId);
String paragraphId = Utils.getParagraphId(jobGroupId);
if (jobUrl != null && noteId != null && paragraphId != null) {
RemoteEventClientWrapper eventClient = ZeppelinContext.getEventClient();
Map<String, String> infos = new java.util.HashMap<>();
@ -187,17 +187,6 @@ public class SparkInterpreter extends Interpreter {
return jobUrl;
}
private String getNoteId(String jobgroupId) {
int indexOf = jobgroupId.indexOf("-");
int secondIndex = jobgroupId.indexOf("-", indexOf + 1);
return jobgroupId.substring(indexOf + 1, secondIndex);
}
private String getParagraphId(String jobgroupId) {
int indexOf = jobgroupId.indexOf("-");
int secondIndex = jobgroupId.indexOf("-", indexOf + 1);
return jobgroupId.substring(secondIndex + 1, jobgroupId.length());
}
};
try {
Object listenerBus = context.getClass().getMethod("listenerBus").invoke(context);
@ -1016,8 +1005,8 @@ public class SparkInterpreter extends Interpreter {
Map<String, String> infos = new java.util.HashMap<>();
if (sparkUrl != null) {
infos.put("url", sparkUrl);
logger.info("Sending metainfos to Zeppelin server: {}", infos.toString());
if (ctx != null && ctx.getClient() != null) {
logger.info("Sending metainfos to Zeppelin server: {}", infos.toString());
getZeppelinContext().setEventClient(ctx.getClient());
ctx.getClient().onMetaInfosReceived(infos);
}

12
spark/src/main/java/org/apache/zeppelin/spark/Utils.java
View file

@ -111,4 +111,16 @@ class Utils {
public static String buildJobGroupId(InterpreterContext context) {
return "zeppelin-" + context.getNoteId() + "-" + context.getParagraphId();
}
public static String getNoteId(String jobgroupId) {
int indexOf = jobgroupId.indexOf("-");
int secondIndex = jobgroupId.indexOf("-", indexOf + 1);
return jobgroupId.substring(indexOf + 1, secondIndex);
}
public static String getParagraphId(String jobgroupId) {
int indexOf = jobgroupId.indexOf("-");
int secondIndex = jobgroupId.indexOf("-", indexOf + 1);
return jobgroupId.substring(secondIndex + 1, jobgroupId.length());
}
}

12
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java
View file

@ -243,7 +243,8 @@ public class RemoteInterpreterEventPoller extends Thread {
Map<String, String> metaInfos = gson.fromJson(event.getData(),
new TypeToken<Map<String, String>>() {
}.getType());
String settingId = getInterpreterSettingId();
String settingId = RemoteInterpreterUtils.
getInterpreterSettingId(interpreterGroup.getId());
listener.onMetaInfosReceived(settingId, metaInfos);
} else if (event.getType() == RemoteInterpreterEventType.PARA_INFOS) {
Map<String, String> paraInfos = gson.fromJson(event.getData(),
@ -251,7 +252,8 @@ public class RemoteInterpreterEventPoller extends Thread {
}.getType());
String noteId = paraInfos.get("noteId");
String paraId = paraInfos.get("paraId");
String settingId = getInterpreterSettingId();
String settingId = RemoteInterpreterUtils.
getInterpreterSettingId(interpreterGroup.getId());
if (noteId != null && paraId != null && settingId != null) {
listener.onParaInfosReceived(noteId, paraId, settingId, paraInfos);
}
@ -342,12 +344,6 @@ public class RemoteInterpreterEventPoller extends Thread {
}
}
private String getInterpreterSettingId() {
String id = interpreterGroup.getId();
int indexOfColon = id.indexOf(":");
return id.substring(0, indexOfColon);
}
private void sendResourcePoolResponseGetAll(ResourceSet resourceSet) {
Client client = null;
boolean broken = false;

10
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java
View file

@ -17,6 +17,7 @@
package org.apache.zeppelin.interpreter.remote;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -63,4 +64,13 @@ public class RemoteInterpreterUtils {
return false;
}
}
public static String getInterpreterSettingId(String intpGrpId) {
String settingId = null;
if (intpGrpId != null) {
int indexOfColon = intpGrpId.indexOf(":");
settingId = intpGrpId.substring(0, indexOfColon);
}
return settingId;
}
}

4
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
View file

@ -2329,7 +2329,7 @@ public class NotebookServer extends WebSocketServlet
metaInfos.remove("paraId");
String label = metaInfos.get("label");
metaInfos.remove("label");
paragraph.updateRuntimeInfos(label, metaInfos, setting.getGroup());
paragraph.updateRuntimeInfos(label, metaInfos, setting.getGroup(), setting.getId());
broadcast(note.getId(), new Message(OP.PARAGRAPH).put("paragraph", paragraph));
}
}
@ -2346,7 +2346,7 @@ public class NotebookServer extends WebSocketServlet
if (note != null) {
Paragraph paragraph = note.getParagraph(paraId);
if (paragraph != null) {
paragraph.clearRuntimeInfo();
paragraph.clearRuntimeInfo(setting.getId());
broadcast(noteId, new Message(OP.PARAGRAPH).put("paragraph", paragraph));
}
}

3
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
View file

@ -379,7 +379,7 @@ public class Note implements Serializable, ParagraphJobListener {
for (Paragraph p : paragraphs) {
if (p.getId().equals(paragraphId)) {
p.setReturn(null, null);
p.clearRuntimeInfo();
p.clearRuntimeInfo(null);
return p;
}
}
@ -564,6 +564,7 @@ public class Note implements Serializable, ParagraphJobListener {
return;
}
p.clearRuntimeInfo(null);
String requiredReplName = p.getRequiredReplName();
Interpreter intp = factory.getInterpreter(p.getUser(), getId(), requiredReplName);

2
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
View file

@ -480,7 +480,7 @@ public class Notebook implements NoteEventListener {
if (p.getDateFinished() != null && lastUpdatedDate.before(p.getDateFinished())) {
lastUpdatedDate = p.getDateFinished();
}
p.clearRuntimeInfo();
p.clearRuntimeInfo(null);
}
Map<String, List<AngularObject>> savedObjects = note.getAngularObjects();

35
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
View file

@ -679,16 +679,17 @@ public class Paragraph extends Job implements Serializable, Cloneable {
}
}
public void updateRuntimeInfos(String label, Map<String, String> infos, String group) {
public void updateRuntimeInfos(String label, Map<String, String> infos, String group,
String intpSettingId) {
if (this.runtimeInfos == null) {
this.runtimeInfos = new HashMap<String, ParagraphRuntimeInfos>();
this.runtimeInfos = new HashMap<String, ParagraphRuntimeInfo>();
}
if (infos != null) {
for (String key : infos.keySet()) {
ParagraphRuntimeInfos info = this.runtimeInfos.get(key);
ParagraphRuntimeInfo info = this.runtimeInfos.get(key);
if (info == null) {
info = new ParagraphRuntimeInfos(key, label, group);
info = new ParagraphRuntimeInfo(key, label, group, intpSettingId);
this.runtimeInfos.put(key, info);
}
info.addValue(infos.get(key));
@ -696,7 +697,29 @@ public class Paragraph extends Job implements Serializable, Cloneable {
}
}
public void clearRuntimeInfo() {
this.runtimeInfos = null;
/**
* Remove runtimeinfo taht were got from the setting with id settingId
* @param settingId
*/
public void clearRuntimeInfo(String settingId) {
if (settingId != null) {
Set<String> keys = runtimeInfos.keySet();
if (keys.size() > 0) {
List<String> infosToRemove = new ArrayList<>();
for (String key : keys) {
ParagraphRuntimeInfo paragraphRuntimeInfo = runtimeInfos.get(key);
if (paragraphRuntimeInfo.getInterpreterSettingId().equals(settingId)) {
infosToRemove.add(key);
}
}
if (infosToRemove.size() > 0) {
for (String info : infosToRemove) {
runtimeInfos.remove(info);
}
}
}
} else {
this.runtimeInfos = null;
}
}
}

37
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/ParagraphRuntimeInfo.java Normal file
View file

@ -0,0 +1,37 @@
package org.apache.zeppelin.notebook;
import java.util.ArrayList;
import java.util.List;
/**
* Store runtime infos of each para
*
*/
public class ParagraphRuntimeInfo {
private String propertyName; //Name of the property
private String label; //Label to be used in UI
private String group; //The interpretergroup from which the info was derived
private List<String> values; // values for the property
private String interpreterSettingId;
public ParagraphRuntimeInfo(String propertyName, String label,
String group, String intpSettingId) {
if (interpreterSettingId == null) {
throw new IllegalArgumentException("Interpreter setting Id cannot be null");
}
this.propertyName = propertyName;
this.label = label;
this.group = group;
this.interpreterSettingId = intpSettingId;
this.values = new ArrayList<>();
}
public void addValue(String value) {
values.add(value);
}
public String getInterpreterSettingId() {
return interpreterSettingId;
}
}

29
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/ParagraphRuntimeInfos.java
View file

@ -1,29 +0,0 @@
package org.apache.zeppelin.notebook;
import java.util.ArrayList;
import java.util.List;
/**
*
*
*/
public class ParagraphRuntimeInfos {
String propertyName;
String label;
String group;
List<String> values;
public ParagraphRuntimeInfos(String propertyName, String label, String group) {
this.propertyName = propertyName;
this.label = label;
this.group = group;
}
public void addValue(String value) {
if (values == null) {
values = new ArrayList<>();
}
values.add(value);
}
}