Spark 中用 Scala 和 java 开发有什么区别

1，构建系统的选择，sbt更合适用来构建Scala工程，maven更合适用来构建Java工程

2，对于spark中的API来说，Java和Scala有差别，但差别并不大

3，如果用Scala开发spark原型程序，可以用spark-shell“打草稿”，或者直接使用spark-shell做交互式实时查询

4，用Scala代码量将减少甚至一个数量级，不过Scala的使用门槛较高

建议：使用Scala构建spark作业，因为spark本身为sbt所构建，同时使用Scala开发spark作业将有助于理解spark的实现机制

简介

由于spark有多种运行模式，远程调试的时候，虽然大体步骤相同，但是还是有小部分需要注意的地方，这里记录一下调试运行在spark on yarn模式下的程序。

环境准备

需要完好的Hadoop，spark集群，以便于提交spark on yarn程序。我这里是基于CDH的环境

步骤

1.随便写个spark程序，比如序列化一个集合，然后求和。然后使用maven打包，上传至集群。可以先提交运行一次，确保可以运行成功。

[root@kjtlxsvr5 bin]# ./spark-submit --class cn.spark.study.core.ParallelizeCollection --master yarn-cluster --num-executors 3 --executor-cores 2 --executor-memory 1G --driver-java-options "-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8787" /home/spark-study-scala-0.0.1-SNAPSHOT-jar-with-dependencies.jar

现在有两个办法可以解决这个问题。

第一个办法是节点少的话，通过修改上面IDEA远程主机地址来一个一个试。

第二办法可以精确知道ApplicationMaster在哪里：

①通过CDH进入yarn的应用程序界面

②然后点击进入该程序的详细信息界面，如下图就可以知道Applicationmaster在哪台NodeManager上：

③可以去该节点查看进程，的确有一个ApplicationMaster，然后在IDEA中修改为该远程主机地址，开始debug程序看源码吧!

package org.jivesoftware.spark.ui;

import java.util.ArrayList;

import org.jivesoftware.spark.util.log.Log;

import org.jivesoftware.spark.util.ModelUtil;

import org.jivesoftware.resource.Res;

import org.jivesoftware.smack.Roster;

import org.jivesoftware.smack.RosterEntry;

import org.jivesoftware.smack.RosterGroup;

import org.jivesoftware.smack.XMPPException;

import org.jivesoftware.smack.packet.Presence;

import org.jivesoftware.smack.packet.RosterPacket;

import org.jivesoftware.smack.util.StringUtils;

import org.jivesoftware.spark.SparkManager;

import org.jivesoftware.spark.UserManager;

/**

* 处理所有订阅请求

* @author jian

*

*/

public class MySubscription {

private String jid;

//构造函数

public MySubscription(){

}

/**

* 调用

* @param jid 主键ID

*/

public void invoke(final String jid){

this.jid=jid;

final Roster roster=SparkManager.getConnection().getRoster();//返回好友册

//如果用户已经在好友当中，则不显示

RosterEntry entry=roster.getEntry(jid);//返回对应的用户

if(entry!=null&&entry.getType()==RosterPacket.ItemType.to){

Presence response=new Presence(Presence.Type.subscribed);

response.setTo(jid);

SparkManager.getConnection().sendPacket(response);

return;

}

UserManager userManager=SparkManager.getUserManager();

String username=userManager.getNickname(userManager.getFullJID(jid));

username=username==null?StringUtils.parseName(UserManager.unescapeJID(jid)):username;//给usernam赋值

boolean addEntry=addEntry(username);

GetGroupName();

if(addEntry){

Presence response=new Presence(Presence.Type.subscribed);

response.setTo(jid);

SparkManager.getConnection().sendPacket(response);

}我爱编程网

}

/**

* 同意添加好友

*/

private boolean addEntry(String username){

String errorMessage=Res.getString("title.error");

String groupName="Friends";

//获取该用户组是否存在

ContactGroup contactGroup=SparkManager.getWorkspace().getContactList().getContactGroup(groupName);

boolean isSharedGroup = contactGroup!=null&&contactGroup.isSharedGroup();

System.out.println("用户组存在?"+isSharedGroup);

if(isSharedGroup){

errorMessage=Res.getString("message.cannot.add.contact.to.shared.group");

}else if(!ModelUtil.hasLength(username)){

errorMessage=Res.getString("message.specify.contact.jid");

}else if(!ModelUtil.hasLength(groupName)){

errorMessage=Res.getString("message.specify.group");

}else if(ModelUtil.hasLength(username)&&ModelUtil.hasLength(groupName)&&!isSharedGroup){

System.out.println("添加好友");

System.out.println("直接输出Group组："+groupName);

addEntity(jid, username, groupName);

return true;

}

return false;

}

/**

* 把新用户添加到好友名单

* @param jid 账号

* @param username 昵称

* @param groupName 组名

*/

private RosterEntry addEntity(String jid,String username,String groupName){

String[] groups={groupName};

System.out.println("开始添加了哦");

for(int i=0;i

System.out.println(groups[i]);

}

Roster roster=SparkManager.getConnection().getRoster();

RosterEntry userEntry=roster.getEntry(jid);

boolean isSubscribed=true;

if(userEntry!=null){

isSubscribed=userEntry.getGroups().size()==0;

}

if(isSubscribed){

try {

roster.createEntry(jid, username, new String[]{groupName});

} catch (XMPPException e) {

Log.error("不能添加新的好友"+jid,e);

}

return roster.getEntry(jid);

}

try {

RosterGroup rosterGroup=roster.getGroup(groupName);

if(rosterGroup==null){

rosterGroup=roster.createGroup(groupName);

}

if(userEntry==null){

roster.createEntry(jid, username, groups);

userEntry=roster.getEntry(jid);

}else {

userEntry.setName(username);

rosterGroup.addEntry(userEntry);

}

userEntry=roster.getEntry(jid);

} catch (XMPPException ex) {

Log.error(ex);

}

return userEntry;

}

private void GetGroupName(){

ArrayList list=new ArrayList();

for(ContactGroup group:SparkManager.getWorkspace().getContactList().getContactGroups()){

if(!group.isOfflineGroup()&&!"Unfiled".equalsIgnoreCase(group.getGroupName())&&!group.isSharedGroup()){

list.add(group.getGroupName());

System.out.println(group.getGroupName());

}

}

}

}