阿里云上面安装hdp服务 hdfs服务，本地 spark stream 消费kafka数据，在hdfs上面设置保存点，但是在写入hdfs的时候报错。
There are 3 datanode(s) running and 3 node(s) are excluded in this operation
解决: 在hdfs-site.xml 中加入此配置参数，使得客户端往访问hdfs返回datanode地址是主机名，
在hosts文件本地配置相对应的映射后 才能访问云主机上面的hadoop 的datanode.
<property>
<name>dfs.client.use.datanode.hostname</name>
<value>true</value>
</property>
020-08-01 11:38:03,938 ERROR --- [ Executor task launch worker for task 80] org.apache.spark.executor.Executor (line: 91) : Exception in task 3.0 in stage 36.0 (TID 80)
org.apache.hadoop.ipc.RemoteException(java.io.IOException):
File /user/atguigu/sparkstreaming/checkpoint
/b7e390a6-0a54-4b67-9401-c9c7eb2bcb6d/rdd-22/.part-00003-attempt-0 could only
be replicated to 0 nodes instead of minReplication (=1).
There are 3 datanode(s) running and 3 node(s) are excluded in this operation.
at org.apache.hadoop.hdfs.server.blockmanagement.BlockManager.chooseTarget4NewBlock(BlockManager.java:1719)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getNewBlockTargets(FSNamesystem.java:3372)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:3296)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:850)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:504)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:640)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:982)
at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2351)
at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2347)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1869)
at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2347)
at org.apache.hadoop.ipc.Client.call(Client.java:1347)
at org.apache.hadoop.ipc.Client.call(Client.java:1300)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:206)
at com.sun.proxy.$Proxy10.addBlock(Unknown Source)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock(ClientNamenodeProtocolTranslatorPB.java:330)
at sun.reflect.GeneratedMethodAccessor65.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:186)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
at com.sun.proxy.$Proxy11.addBlock(Unknown Source)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.locateFollowingBlock(DFSOutputStream.java:1226)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.nextBlockOutputStream(DFSOutputStream.java:1078)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:514)
import java.lang
import java.sql.ResultSet
import com.atguigu.qzpoint.util.{DataSourceUtil, QueryCallback, SqlProxy}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.mutable
import scala.util.Random
object RegisterStreaming {
private val groupid = "register_group11"
def main(args: Array[String]): Unit = {
// System.setProperty("HADOOP_USER_NAME", "root")
val conf = new SparkConf().setAppName(this.getClass.getSimpleName)
.set("spark.streaming.kafka.maxRatePerPartition", "100")
// .set("spark.streaming.backpressure.enabled", "true")
// .set("spark.streaming.stopGracefullyOnShutdown", "true")
.setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(3))
val sparkContext: SparkContext = ssc.sparkContext
sparkContext.hadoopConfiguration.set("fs.defaultFS", "hdfs://hadoopha1")
sparkContext.hadoopConfiguration.set("dfs.nameservices", "hadoopha1")
val topics = Array("register_topic")
val kafkaMap: Map[String, Object] = Map[String, Object](
"bootstrap.servers" -> "hadoop102:6667,hadoop103:6667,hadoop104:6667",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> groupid,
"auto.offset.reset" -> "earliest", //sparkstreaming第一次启动，不丢数
//如果是true，则这个消费者的偏移量会在后台自动提交，但是kafka宕机容易丢失数据
//如果是false，则需要手动维护kafka偏移量
"enable.auto.commit" -> (false: lang.Boolean)
)
//sparkStreaming对有状态的数据操作，需要设定检查点目录，然后将状态保存到检查点中
ssc.checkpoint("/user/atguigu/sparkstreaming/checkpoint")
//查询mysql中是否有偏移量
val sqlProxy = new SqlProxy()
val offsetMap = new mutable.HashMap[TopicPartition, Long]()
val client = DataSourceUtil.getConnection
try {
sqlProxy.executeQuery(client, "select * from `offset_manager` where groupid=?", Array(groupid), new QueryCallback {
override def process(rs: ResultSet): Unit = {
while (rs.next()) {
val model = new TopicPartition(rs.getString(2), rs.getInt(3))
val offset = rs.getLong(4)
offsetMap.put(model, offset)
}
rs.close() //关闭游标
}
})
} catch {
case e: Exception => e.printStackTrace()
} finally {
sqlProxy.shutdown(client)
}
//设置kafka消费数据的参数 判断本地是否有偏移量 有则根据偏移量继续消费 无则重新消费
val stream: InputDStream[ConsumerRecord[String, String]] = if (offsetMap.isEmpty) {
KafkaUtils.createDirectStream(
ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap))
} else {
KafkaUtils.createDirectStream(
ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap, offsetMap))
}
//stream原始流无法进行使用和打印，会报序列化错误，所以需要做下面的map转换
val resultDStream = stream.filter(item => item.value().split("\t").length == 3).
mapPartitions(partitions => {
partitions.map(item => {
val line = item.value()
val arr = line.split("\t")
val app_name = arr(1) match {
case "1" => "PC"
case "2" => "APP"
case _ => "Other"
}
(app_name, 1)
})
})
resultDStream.cache()
//(PC,1),(PC,1),(APP,1),(Other,1),(APP,1),(Other,1),(PC,1),(APP,1)
//"=================每6s间隔1分钟内的注册数据================="
// resultDStream.reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(60), Seconds(6)).print()
//"========================================================="
//"+++++++++++++++++++++++实时注册人数+++++++++++++++++++++++"//状态计算
val updateFunc = (values: Seq[Int], state: Option[Int]) => {
val currentCount = values.sum //本批次求和
val previousCount = state.getOrElse(0) //历史数据
Some(currentCount + previousCount)
}
resultDStream.updateStateByKey(updateFunc).print()
//"++++++++++++++++++++++++++++++++++++++++++++++++++++++++"
/* val dsStream = stream.filter(item => item.value().split("\t").length == 3)
.mapPartitions(partitions =>
partitions.map(item => {
val rand = new Random()
val line = item.value()
val arr = line.split("\t")
val app_id = arr(1)
(rand.nextInt(3) + "_" + app_id, 1)
}))
val result = dsStream.reduceByKey(_ + _)
result.map(item => {
val appid = item._1.split("_")(1)
(appid, item._2)
}).reduceByKey(_ + _).print()*/
//处理完 业务逻辑后 手动提交offset维护到本地 mysql中
stream.foreachRDD(rdd => {
val sqlProxy = new SqlProxy()
val client = DataSourceUtil.getConnection
try {
val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
for (or <- offsetRanges) {
sqlProxy.executeUpdate(client, "replace into `offset_manager` (groupid,topic,`partition`,untilOffset) values(?,?,?,?)",
Array(groupid, or.topic, or.partition.toString, or.untilOffset))
}
} catch {
case e: Exception => e.printStackTrace()
} finally {
sqlProxy.shutdown(client)
}
})
ssc.start()
ssc.awaitTermination()
}
}
