UqrkOCyfkQZc
Causedby:java.lang.ClassNotFoundException:org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactory 原来的导包依赖是： <groupId>org.apache.flink</groupId> <artifactId>flink-connector-hive_2.12</artifactId> <version>${flink.version}</version> </dependency>...
这个花了4天时间才解决了win本地flink连接hivecatalog的目标 1.添加jar依赖包 2.flinksql验证catalog的注册 3.flink_home/lib下的flink-table-planner-loader_2.12-1.16.0.jar与flink_home/opt下的flink-table-planner_2.12-1.16.0.jar互换 4.flink1.16中已经去掉了$FLIN_HOME/conf/sql-client-default.yaml的设置，直接在flinksql客户端中配置； 5.主要原因是引入的包太多了，导致包冲突，可以把类似名字的包先注释...
flink1.16.0 hive-2.3.9 hadoop2.7.7（本地安装） packagecn.mrt.flink.connector; importorg.apache.flink.table.api.EnvironmentSettings; importorg.apache.flink.table.api.SqlDialect; importorg.apache.flink.table.api.TableEnvironment; importorg.apache.flink.table.catalog.hive.HiveCatalog; publicclassHiveCon...
sql逻辑： createtableods.product(product_idint,product_namevarchar(255))ENGINE=InnoDBDEFAULTCHARSET=utf8mb3; insertintoods.productvalues(1,'LCPhone'); insertintoods.productvalues(2,'LCT-Shirt'); selectfromods.product; createtableods.sales(product_idint,product_startdate,product_enddate,average_dai...
2023-04-1716:09:52,379INFO[org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager]Receivedresourcerequirementsfromjoba0cd5552d863fdcb12909631610e0c54:[ResourceRequirement{resourceProfile=ResourceProfile{UNKNOWN},numberOfRequiredSlots=1}] 2023-04-1716:09:52,416ERROR[org.apache.f...
tableEnv.executeSql("createtablemysql_binlog(\n"+ "idSTRINGnotnull,\n"+ "region_nameSTRING,\n"+ "primarykey(id)NOTENFORCED\n"+//两个条件缺一不可 ")with(\n"+ "'connector'='mysql-cdc',\n"+ "'hostname'='192.168.56.131',\n"+ "'port'='3306',\n"+ "'username'='root',\n"+ "'password'='root123456',\n"+ "'database-n...
.java.lang.VerifyError:classcom.google.protobuf.LiteralByteStringoverridesfinalmethodtoString.(Ljava/lang/String;)Ljava/lang/String; 找到关键字protobuf，猜测的原因是hbase的依赖包冲突，于是先将hbase里的包排除，再降低版本引入 <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java<...
之前一直以为需要配置特别的参数来让flink按照sink事务发往不同kafka，参照了不同版本的写法，flink1.13还是使用flinkKafkaproducer接口，flink1.16已经过时，但是还能使用，并且引进新的接口kafkasink.而本次的接口并非是接口的问题，而是令人意想不到的数据源mysql，mysql数据源之前是通过mysql.init脚本初始化，所以以为所有的表都有数据，没有去检查，直到最新想换个思路，看是否跟表的顺序有关，结果发现没有关系，怀疑是表的问题。才去检查表的数据。浪费了2天时间。哎！！！ MySqlSource<String>mySqlSour...
之前一开始的代码是这样： Propertiesproperties=newProperties(); properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,Constants.KAFKA_BROKER); properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringDeserializer.class.getName()); properties.setPro...
问题1：没有重写方法toString 最终的合流无法显示，是因为orderWide没有重写方法toString 并且此合流后的数据还没有sink，仅仅是消费kafka，因此重写方法后,默认之前的kafkaoffset已经提交，对于同一消费组不会再重复消费，所以需要修改消费组再重新运行。 @Override publicStringtoString(){ returnJSON.toJSONString(this,SerializerFeature.WriteMapNullValue); } 问题2：如何将json中null字段打印输出，可以不区分字段类型，所有null都输出为"...
背景： 在ods->dwd时，由于需要同时对pojo类型的kafka多流（消费多个topic）数据消费时需要操作类似的反序列化，所以需要先将此操泛型化。加入class这个参数。即将操作对象泛型化。 原来的分topic反序列化： KafkaSource<OrderInfo>orderInfoKafkaSource=KafkaSource.<OrderInfo>builder() .setBootstrapServers(Constants.KAFKA_BROKER) .setTopics(orderInfoSourceTopic) .setGroupId(grou...
表现的现象是hbase进程HMaster启动后，几秒后就退出。然后看前后的报错似乎很hdfs有关，于是登录host:50070查看发现有3个block不见了，网上的说法有认为是hbase元数据损坏，有认为是hbase写入数据过多，总而言之，需要删除hdfs上hbase的文件夹。然后重启hbase,zookeeper，这个方法明显的缺陷是之前hbase上的数据没有备份，正确的方式一定是先备份data数据。 [hadoop@hadoop]$hdfsdfs-rm-r/hbase [hadoop@hadoop]$zkCli.sh [zk:localhost:2181(CONNECTED)0]del...
这个主要是jdbc的依赖问题 ,newJdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(Constants.CLICKHOUSE_URL) .withDriverName(Constants.CLICKHOUSE_DRIVER) .build() publicstaticfinalStringCLICKHOUSE_URL="jdbc:clickhouse://hadoop:8123/default"; //publicstaticfinalStringCLICKHOUSE_DRIVER="com.clickhous...
"select"+ "DATE_FORMAT(TUMBLE_START(rt,INTERVAL'10'SECOND),'yyyy-MM-ddHH:mm:ss')stt,"+ "DATE_FORMAT(TUMBLE_END(rt,INTERVAL'10'SECOND),'yyyy-MM-ddHH:mm:ss')edt,"+ "province_id,"+ "province_name,"+ "province_area_code,"+ "province_iso_code,"+ "count(distinctorder_id)order_count,"+ "sum(original_total...
前提读取的文件已经put到hdfs上了，还是报错，仔细想想，为什么两个读取文件只报后面那个读取文件不存在呢？看代码，是读取的方式不同，前面一个是通过sparkcontext读取，后面是file,所以情况应该是只有通过spark生成的对象sc读取才可以，带着这个思路，修改代码，才运行成功。 JavaRDD<String>linesRDD2=sc.textFile("src/main/resources/santi/bad_words.txt"); //JavaRDD<String>linesRDD2=sc.textFile("/tmp/bad_words.txt");...
nohupzkServer.shstart$ZOOKEEPER_HOME/conf/zoo.cfg& nohupkafka-server-start.sh$KAFKA_HOME/config/server.properties& 一直报错，进程jps闪退 排查过程：去kafka安装路径下的log/server.log中定位kafka启动过程报错原因 解决方案： 此次的kakfa闪退有两个原因：一个原因是kafka的dir.log路径不能跟kafka的运行日志路径相同，否则报错： ERROR[KafkaServerid=0]FatalerrorduringKafkaSe...