1

Kafka-> Apache Apex->の実行中Hbaseに、Yarn タスクで次の例外が発生しています。

com.datatorrent.stram.StreamingAppMasterService: Application master, appId=4, clustertimestamp=1479188884109, attemptId=2
2016-11-15 11:59:51,068 INFO org.apache.hadoop.service.AbstractService: Service com.datatorrent.stram.StreamingAppMasterService failed in state INITED; cause: java.lang.RuntimeException: java.lang.reflect.InvocationTargetException
java.lang.RuntimeException: java.lang.reflect.InvocationTargetException
    at org.apache.hadoop.fs.AbstractFileSystem.newInstance(AbstractFileSystem.java:130)
    at org.apache.hadoop.fs.AbstractFileSystem.createFileSystem(AbstractFileSystem.java:156)
    at org.apache.hadoop.fs.AbstractFileSystem.get(AbstractFileSystem.java:241)
    at org.apache.hadoop.fs.FileContext$2.run(FileContext.java:333)
    at org.apache.hadoop.fs.FileContext$2.run(FileContext.java:330)
    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:1614)
    at org.apache.hadoop.fs.FileContext.getAbstractFileSystem(FileContext.java:330)
    at org.apache.hadoop.fs.FileContext.getFileContext(FileContext.java:444)

そして、次の例外を示す DataTorrent ログ。Kafka -> Apex -> Hbase ストリーミング アプリケーションを通信するアプリを実行しています。

 Connecting to ResourceManager at hduser1/127.0.0.1:8032
 16/11/15 17:47:38 WARN client.EventsAgent: Cannot read events for application_1479208737206_0008: java.io.FileNotFoundException: File does not exist: /user/hduser1/datatorrent/apps/application_1479208737206_0008/events/index.txt
at org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:66)
at org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:56)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocationsUpdateTimes(FSNamesystem.java:1893)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocationsInt(FSNamesystem.java:1834)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocations(FSNamesystem.java:1814)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getBlockLocations(FSNamesystem.java:1786)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.getBlockLocations(NameNodeRpcServer.java:552)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.getBlockLocations(ClientNamenodeProtocolServerSideTranslatorPB.java:362)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:619)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:962)
at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2040)
at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2036)
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:1656)
at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2034)

コードの追加:

public void populateDAG(DAG dag, Configuration conf){
KafkaSinglePortInputOperator in
  = dag.addOperator("kafkaIn", new KafkaSinglePortInputOperator());

in.setInitialOffset(AbstractKafkaInputOperator.InitialOffset.EARLIEST.name());
LineOutputOperator out = dag.addOperator("fileOut", new LineOutputOperator());

dag.addStream("data", in.outputPort, out.input);}

LineOutputOperator は AbstractFileOutputOperator を拡張します

private static final String NL = System.lineSeparator();
private static final Charset CS = StandardCharsets.UTF_8;

@NotNull
private String baseName;

@Override
public byte[] getBytesForTuple(byte[] t) {
  String result = new String(t, CS) + NL;
  return result.getBytes(CS);
}

@Override
protected String getFileName(byte[] tuple) {
 return baseName;
}

public String getBaseName() { return baseName; }
public void setBaseName(String v) { baseName = v; }

この問題を解決するにはどうすればよいですか?

ありがとう。

4

2 に答える 2

1

Hadoop や apex のバージョンなど、環境に関する詳細を教えてください。また、この例外はどのログに表示されますか?

簡単なサニティ チェックとして、http: //docs.datatorrent.com/beginner/で説明されているシンプルな Maven アーキタイプ生成アプリケーションを実行できますか?

それが機能する場合は、 https ://github.com/DataTorrent/examples/tree/master/tutorials で fileIO および kafka アプリケーションを実行してみて ください。

それらが正常に機能する場合は、コードの詳細を確認できます。

于 2016-11-15T16:13:49.903 に答える
0

私は解決策を得ました、

私のライセンスの有効期限に関連する問題なので、新しいライセンスを再インストールし、実際のコードでは問題なく動作します。

于 2016-11-22T06:32:24.460 に答える