使用 Spark SQL 在 cassandra 中加入两个表 - 错误：缺少 EOF

Posted 2023-04-17

【中文标题】使用 Spark SQL 在 cassandra 中加入两个表 - 错误：缺少 EOF

【英文标题】：using SparkSQL to join two tables on cassandra - ERROR: missing EOF

【发布时间】：2016-03-30 09:21:45

【问题描述】：

我在我的机器上安装了带有 SparkSQL 的 Cassandra 和 Spark。 Spark SQL 支持 JOIN 关键字

https://docs.datastax.com/en/datastax_enterprise/4.6/datastax_enterprise/spark/sparkSqlSupportedSyntax.html

Spark SQL 支持的语法 以下语法定义了一个 SELECT 查询。

SELECT [DISTINCT] [列名]|[通配符] FROM [kesypace name.] 表名 [JOIN 子句表名 ON 连接条件] [WHERE 条件] [GROUP BY 列名] [HAVING 条件] [ORDER BY 列 名称 [ASC | DSC]]

我有以下代码

SparkConf conf = new SparkConf().setAppName("My application").setMaster("local");
conf.set("spark.cassandra.connection.host", "localhost");
JavaSparkContext sc = new JavaSparkContext(conf);
CassandraConnector connector = CassandraConnector.apply(sc.getConf());
Session session = connector.openSession();

ResultSet results;
String sql ="";


BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
sql = "SELECT * from siem.report JOIN siem.netstat on siem.report.REPORTUUID = siem.netstat.NETSTATREPORTUUID ALLOW FILTERING;";
results = session.execute(sql);

我收到以下错误

原因：com.datastax.driver.core.exceptions.SyntaxError: line 1:25 ',' 处缺少 EOF（SELECT * from siem.report[,] siem...）上午 11:14 com.datastax.driver.core.exceptions.SyntaxError：第 1:25 行缺少 EOF 在 ',' (SELECT * from siem.report[,] siem...) 在 com.datastax.driver.core.exceptions.SyntaxError.copy(SyntaxError.java:58) 在 com.datastax.driver.core.exceptions.SyntaxError.copy(SyntaxError.java:24) 在 com.datastax.driver.core.DriverThrowables.propagateCause（DriverThrowables.java:37） 在 com.datastax.driver.core.DefaultResultSetFuture.getUninterruptibly(DefaultResultSetFuture.java:245) 在 com.datastax.driver.core.AbstractSession.execute(AbstractSession.java:63) 在 com.datastax.driver.core.AbstractSession.execute(AbstractSession.java:39) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:483) 在 com.datastax.spark.connector.cql.SessionProxy.invoke(SessionProxy.scala:33) 在 com.sun.proxy.$Proxy59.execute（未知来源） com.ge.predix.rmd.siem.boot.PersistenceTest.test_QuerySparkOnReport_GIACOMO_LogDao(PersistenceTest.java:178) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:483) 在 org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50) 在 org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) 在 org.junit.runners.model.FrameworkMethod.invokeExplosively（FrameworkMethod.java:47） 在 org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) 在 org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:73) 在 org.springframework.test.context.junit4.statements

也试过

SELECT * from siem.report JOIN siem.netstat on report.REPORTUUID = netstat.NETSTATREPORTUUID ALLOW FILTERING

也试过

SELECT * from siem.report R JOIN siem.netstat N on R.REPORTUUID = N.NETSTATREPORTUUID ALLOW FILTERING

有人可以帮助我吗？我真的在使用 SparkSQL 或 CQL？

更新

我试过了

public void test_JOIN_on_Cassandra () 

        SparkConf conf = new SparkConf().setAppName("My application").setMaster("local");
        conf.set("spark.cassandra.connection.host", "localhost");
        JavaSparkContext sc = new JavaSparkContext(conf);


        SQLContext sqlContext = new SQLContext(sc);
        try 
            //QueryExecution test1 = sqlContext.executeSql("SELECT * from siem.report");
            //QueryExecution test2 = sqlContext.executeSql("SELECT * from siem.report JOIN siem.netstat on report.REPORTUUID = netstat.NETSTATREPORTUUID");
            QueryExecution test3 = sqlContext.executeSql("SELECT * from siem.report JOIN siem.netstat on siem.report.REPORTUUID = siem.netstat.NETSTATREPORTUUID");

         catch (Exception e) 
            e.printStackTrace();
        

       // SchemaRDD results = sc.sql("SELECT * from siem.report JOIN siem.netstat on siem.report.REPORTUUID = siem.netstat.NETSTATREPORTUUID");

我明白了

== 已解析的逻辑计划 == '项目 [unresolvedalias()] +- '加入内部，Some(('siem.report.REPORTUUID = 'siem.netstat.NETSTATREPORTUUID)) :- 'UnresolvedRelation siem.report, 无 +- 'UnresolvedRelation siem.netstat, 无 == 分析的逻辑计划 == org.apache.spark.sql.catalyst.analysis.UnresolvedException: Invalid 在未解析的对象上调用 toAttribute，树：unresolvedalias() '项目[未解决的别名（*）] +- '加入内部，Some(('siem.report.REPORTUUID = 'siem.netstat.NETSTATREPORTUUID)) :- 'UnresolvedRelation siem.report, 无 +- 'UnresolvedRelation siem.netstat, 无 == 优化逻辑计划 == org.apache.spark.sql.AnalysisException: 找不到表：siem.report; == 物理计划 == org.apache.spark.sql.AnalysisException：找不到表：siem.report;

【参考方案1】：

您似乎在这里混合了几个导致错误的概念。您正在创建的会话将打开一条通往 Cassandra 的直接线路，这意味着它将接受 CQL 而不是 SQL。如果你想运行 SQL，你可以做一个小改动

SparkConf conf = new SparkConf().setAppName("My application").setMaster("local");
conf.set("spark.cassandra.connection.host", "localhost");
JavaSparkContext sc = new JavaSparkContext(conf);

SchemaRDD results = sparkContext.sql("SELECT * from siem.report JOIN siem.netstat on siem.report.REPORTUUID = siem.netstat.NETSTATREPORTUUID");

您从 Spark 上下文调用 SparkSQL，而不是直接连接到 Cassandra。更多内容：http://docs.datastax.com/en/latest-dse/datastax_enterprise/spark/sparkSqlJava.html