首页 文章 精选 留言 我的

精选列表

搜索[信息窗口],共10000篇文章
优秀的个人博客,低调大师

4月末阿里云ACE同城会活动信息一览

从本周起,我们会每周发布未来1周左右阿里云ACE同城会线下活动一览,大家可根据自己的时间安排、地点报名参加。 活动一主题:30分钟学会搭建高可用网站时间:4月27日下午14:00~17:00地点:厦门市思明区软件园二期望海路57号之一501实验介绍网站建设是一项复杂的工程,随着规模的扩大,许多网站势必会遇到严重的可用性问题。大量用户涌入时如何保证网站不崩溃?这是摆在网站开发和运维人员面前迫待解决的问题。本实验通过使用阿里云常用的四个云产品,来帮助用户快速地在云端搭建具备高可用性的网站。由于阿里云的云产品提供了很高的数据可靠性和业务可用性,从而保证部署在云端上的网站在不同极端情况下都能够正常地运行。报名地址https://mvp.aliyun.com/crowdsource/assignment/335 活动二主题:GreenTea J

优秀的个人博客,低调大师

K8S集群中修改在线应用的存储卷信息

修改应用NAS版本:从v4到v3 1. 修改nfs版本,需要重挂载nas卷,则pod需要重启; 2. 如果使用deploy部署应用,参考下面示例,如果使用pod部署应用,下面示例中改为pod的部署模板; 修改前应用: pv.yaml apiVersion: v1 kind: PersistentVolume metadata: name: pv-nas labels: alicloud-pvname: pv-nas spec: capacity: storage: 5Gi accessModes: - ReadWriteMany flexVolume: driver: "alicloud/nas" options: server: "2564f49129-ggu23.

优秀的个人博客,低调大师

关于flink消费kafka的序列化时-如何拿到消息的meta信息

首先,我们故意制造一个异常 然后,发送一个JSON数据,并且缺失了一些字段,看看报什么错! 顺利抓到调用栈 [2018-11-23 13:24:32,877] INFO Source: MyKafka010JsonTableSource -> from: (pro, throwable, level, ip, SPT) -> where: (AND(IS NOT NULL(pro), IS NOT NULL(throwable), IS NOT NULL(level), IS NOT NULL(ip))), select: (pro, throwable, level, ip, SPT) (3/3) (7da5fcb7cd77ff35b64d8bbe367a84d7) switched from RUNNING to FAILED. org.apache.flink.runtime.taskmanager.Task.transitionState(Task.java:927) java.io.IOException: Failed to deserialize JSON object. at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:97) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:50) at org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper.deserialize(KeyedDeserializationSchemaWrapper.java:44) at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:142) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:721) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.IllegalStateException: Could not find field with name 'ip'. at org.apache.flink.formats.json.JsonRowDeserializationSchema.convertRow(JsonRowDeserializationSchema.java:189) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:95) ... 10 more [2018-11-23 13:24:32,877] INFO Source: MyKafka010JsonTableSource -> from: (pro, throwable, level, ip, SPT) -> where: (AND(IS NOT NULL(pro), IS NOT NULL(throwable), IS NOT NULL(level), IS NOT NULL(ip))), select: (pro, throwable, level, ip, SPT) (2/3) (f68577aede6953de859c636d9d3b6f3e) switched from RUNNING to FAILED. org.apache.flink.runtime.taskmanager.Task.transitionState(Task.java:927) java.io.IOException: Failed to deserialize JSON object. at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:97) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:50) at org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper.deserialize(KeyedDeserializationSchemaWrapper.java:44) at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:142) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:721) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.IllegalStateException: Could not find field with name 'ip'. at org.apache.flink.formats.json.JsonRowDeserializationSchema.convertRow(JsonRowDeserializationSchema.java:189) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:95) ... 10 more [2018-11-23 13:24:32,877] INFO Source: MyKafka010JsonTableSource -> from: (pro, throwable, level, ip, SPT) -> where: (AND(IS NOT NULL(pro), IS NOT NULL(throwable), IS NOT NULL(level), IS NOT NULL(ip))), select: (pro, throwable, level, ip, SPT) (1/3) (8a2c89aa3a2ea8f4c4423c69b22b2522) switched from RUNNING to FAILED. org.apache.flink.runtime.taskmanager.Task.transitionState(Task.java:927) java.io.IOException: Failed to deserialize JSON object. at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:97) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:50) at org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper.deserialize(KeyedDeserializationSchemaWrapper.java:44) at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:142) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:721) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.IllegalStateException: Could not find field with name 'ip'. at org.apache.flink.formats.json.JsonRowDeserializationSchema.convertRow(JsonRowDeserializationSchema.java:189) at org.apache.flink.formats.json.JsonRowDeserializationSchema.deserialize(JsonRowDeserializationSchema.java:95) ... 10 more [2018-11-23 13:24:32,892] INFO Freeing task resources for Source: MyKafka010JsonTableSource -> from: (pro, throwable, level, ip, SPT) -> where: (AND(IS NOT NULL(pro), IS NOT NULL(throwable), IS NOT NULL(level), IS NOT NULL(ip))), select: (pro, throwable, level, ip, SPT) (1/3) (8a2c89aa3a2ea8f4c4423c69b22b2522). org.apache.flink.runtime.taskmanager.Task.run(Task.java:810) 这里只是为了让大家对内部序列化的地点有个印象。 如果你看懂了,下面进入正题 1)自定义的TableSource中改写 protected FlinkKafkaConsumerBase<Row> createKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) { //下面2行本质是一样的,但是因为需要自定义,所以自己返回 //super.createKafkaConsumer(topic, properties, deserializationSchema); return new FlinkKafkaConsumer010<>(Collections.singletonList(topic), new MyKeyedDeserializationSchemaWrapper<>(deserializationSchema), properties); } 2)MyKeyedDeserializationSchemaWrapper.java package org.apache.flink.api.java.io.kafka; import java.io.IOException; import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.java.typeutils.RowTypeInfo; import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper; import org.apache.flink.types.Row; public class MyKeyedDeserializationSchemaWrapper<T> extends KeyedDeserializationSchemaWrapper<T> { // private static final org.slf4j.Logger Logger = LoggerFactory // .getLogger(MyKeyedDeserializationSchemaWrapper.class); private static final int NOT_FOUND = -1; /** */ private static final long serialVersionUID = -6984024233082983820L; // private static final String KAFKA_TOPIC = "KAFKA_TOPIC"; private int KAFKA_TOPIC_INDEX = NOT_FOUND; // private static final String KAFKA_PARTITION = "KAFKA_PARTITION"; private int KAFKA_PARTITION_INDEX = NOT_FOUND; // private static final String KAFKA_OFFSET = "KAFKA_OFFSET"; private int KAFKA_OFFSET_INDEX = NOT_FOUND; // private boolean inited = false; public MyKeyedDeserializationSchemaWrapper(DeserializationSchema<T> deserializationSchema) { super(deserializationSchema); } @SuppressWarnings("unchecked") @Override public T deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException { Row row = (Row) super.deserialize(messageKey, message, topic, partition, offset); //只执行1次 if (false == inited) { RowTypeInfo rowTypeInfo = (RowTypeInfo) super.getProducedType();//Row(pro: String, throwable: String, level: String, ip: String, SPT: Timestamp) KAFKA_TOPIC_INDEX = rowTypeInfo.getFieldIndex(KAFKA_TOPIC); KAFKA_PARTITION_INDEX = rowTypeInfo.getFieldIndex(KAFKA_PARTITION); KAFKA_OFFSET_INDEX = rowTypeInfo.getFieldIndex(KAFKA_OFFSET); inited = true; } //每次都要执行一遍 if (NOT_FOUND != KAFKA_TOPIC_INDEX) { row.setField(KAFKA_TOPIC_INDEX, topic); } if (NOT_FOUND != KAFKA_PARTITION_INDEX) { row.setField(KAFKA_PARTITION_INDEX, partition); } if (NOT_FOUND != KAFKA_OFFSET_INDEX) { row.setField(KAFKA_OFFSET_INDEX, offset); } //返回最终结果 return (T) row; } } 测试一把看看效果! 完美!

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册