Beam安装：
https://beam.apache.org/get-started/quickstart-py/

为安装Apache Beam，首先下载安装了最新版python 3.12.4。

> python --version
Python 3.12.4

按文档安装beam时报错无法编译wheel：

> pip install apache-beam
...
      INFO:root:copying apache_beam\portability\api\org\apache\beam\model\pipeline\v1\schema_pb2.pyi -> build\lib.win-amd64-cpython-312\apache_beam\portability\api\org\apache\beam\model\pipeline\v1
      INFO:root:copying apache_beam\portability\api\org\apache\beam\model\pipeline\v1\standard_window_fns_pb2.pyi -> build\lib.win-amd64-cpython-312\apache_beam\portability\api\org\apache\beam\model\pipeline\v1
      INFO:root:copying apache_beam\portability\api\standard_coders.yaml -> build\lib.win-amd64-cpython-312\apache_beam\portability\api
      INFO:root:running build_ext
      INFO:root:building 'apache_beam.coders.coder_impl_row_encoders' extension
      error: Microsoft Visual C++ 14.0 or greater is required. Get it with "Microsoft C++ Build Tools": https://visualstudio.microsoft.com/visual-cpp-build-tools/
      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.
  ERROR: Failed building wheel for apache-beam
  Building wheel for pyarrow (pyproject.toml) ... error
  error: subprocess-exited-with-error

  × Building wheel for pyarrow (pyproject.toml) did not run successfully.
  │ exit code: 1
  ╰─> [299 lines of output]
      <string>:34: DeprecationWarning: pkg_resources is deprecated as an API. See https://setuptools.pypa.io/en/latest/pkg_resources.html
      WARNING setuptools_scm.pyproject_reading toml section missing 'pyproject.toml does not contain a tool.setuptools_scm section'
      Traceback (most recent call last):
        File "C:\Users\zhanghao\AppData\Local\Temp\pip-build-env-0wvk6red\overlay\Lib\site-packages\setuptools_scm\_integration\pyproject_reading.py", line 36, in read_pyproject
          section = defn.get("tool", {})[tool_name]
                    ~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^
      KeyError: 'setuptools_scm'
      running bdist_wheel
      running build
      running build_py

按报错信息里的提示，从 https://visualstudio.microsoft.com/zh-hans/visual-cpp-build-tools/ 下载了最新版的vc++ buildtools仍然报错。

在so上找到类似的问题，其中一个解决方法是安装整个vc++，但需要下载6GB+太重了。发现还有另一个方法，就是不要使用最新版的python，因为最新版python里没有包含一些已编译的包。

https://stackoverflow.com/questions/64261546/how-to-solve-error-microsoft-visual-c-14-0-or-greater-is-required-when-inst

This error can happen when using the latest version of Python, e.g. 3.12, because the package wheels were only built for earlier versions of Python

卸载python 3.12.4，安装python 3.9.13，再重新安装beam成功。

> python --version
Python 3.9.13

> pip install apache-beam
...
Using legacy 'setup.py install' for crcmod, since package 'wheel' is not installed.
Using legacy 'setup.py install' for dill, since package 'wheel' is not installed.
Using legacy 'setup.py install' for hdfs, since package 'wheel' is not installed.
Using legacy 'setup.py install' for pyjsparser, since package 'wheel' is not installed.
Using legacy 'setup.py install' for docopt, since package 'wheel' is not installed.
Installing collected packages: pytz, pyjsparser, docopt, crcmod, zstandard, urllib3, tzdata, typing-extensions, six, rpds-py, regex, pyparsing, pyarrow-hotfix, protobuf, packaging, orjson, objsize, numpy, jsonpickle, idna, grpcio, fasteners, fastavro, dnspython, dill, cloudpickle, charset-normalizer, certifi, attrs, async-timeout, tzlocal, requests, referencing, redis, python-dateutil, pymongo, pydot, pyarrow, proto-plus, httplib2, jsonschema-specifications, js2py, hdfs, jsonschema, apache-beam
  Running setup.py install for pyjsparser ... done
  Running setup.py install for docopt ... done
  Running setup.py install for crcmod ... done
  Running setup.py install for dill ... done
  Running setup.py install for hdfs ... done
Successfully installed apache-beam-2.56.0 async-timeout-4.0.3 attrs-23.2.0 ...

这篇是速记，基于TDengine 2.6版，详情建议看官方文档：数据节点管理

TDengine集群版通过虚拟节点（vgroup）实现database的负载均衡和水平扩展，每个物理节点(dnode）可以包含多个vgroup。每个database由多个table组成，这些table根据创建顺序被分布到不同的vgroup。

database1
    vgroup1
        table1
        table2
        ...
        table10
    vgroup2
        table11
        ...
        table20

table分布到不同vgroup的规则是由taos.cfg里的minTablesPerVnode参数和tableIncStepPerVnode参数控制的，每当database里当前vnode的table数量超过minTablesPerVnode时就会创建一个新的vgroup。这两个参数的默认值很大，如果database里table的数量不多，建议手工配置减小参数值，否则可能所有数据都分配到同一个物理节点产生数据倾斜。

注意：这两个参数修改后需要重启taosd服务，已经创建好的database需要删除重建才会生效。

[root@taos-1 ~]# tail /etc/taos/taos.cfg
...
minTablesPerVnode                 10
tableIncStepPerVnode              10

注意：在tdengine 3.0里minTablesPerVnode和tableIncStepPerVnode这两个参数已经被移除。

TDengine允许指定vgroup的副本数，每个副本是一个vnode。如果vgroup的副本数大于1，则此vgroup里会包含多个vnode，不同的vnode会被分配到不同的物理节点上。

可以在taos命令行里查看vgroups的情况。如下面的命令可以看到，集群包含3个物理节点，数据库里共有80个表，每10个表共用一个vgroup，这些vgroup比较均匀的分布到3个物理节点上。

[root@taos-1 ~]# taos
taos> use database1;
taos> show dnodes;
   id   |           end_point            | vnodes | cores  |   status   | role  |       create_time       |      offline reason      |
======================================================================================================================================
      1 | taos-1:6030                    |     61 |      8 | ready      | any   | 2022-04-03 17:58:10.049 |                          |
      2 | taos-2:6030                    |     61 |      8 | ready      | any   | 2022-04-03 17:58:15.756 |                          |
      3 | taos-3:6030                    |     62 |      8 | ready      | any   | 2022-02-03 17:58:16.287 |                          |
Query OK, 3 row(s) in set (0.008325s)

taos> show vgroups;
    vgId     |   tables    |  status  |   onlines   | v1_dnode | v1_status | compacting  |
==========================================================================================
         536 |          10 | ready    |           1 |        3 | leader    |           0 |
         537 |          10 | ready    |           1 |        2 | leader    |           0 |
         538 |          10 | ready    |           1 |        1 | leader    |           0 |
         539 |          10 | ready    |           1 |        3 | leader    |           0 |
         540 |          10 | ready    |           1 |        2 | leader    |           0 |
         541 |          10 | ready    |           1 |        1 | leader    |           0 |
         542 |          10 | ready    |           1 |        3 | leader    |           0 |
         543 |          10 | ready    |           1 |        2 | leader    |           0 |
Query OK, 8 row(s) in set (0.005023s)

公司产品为支持流式数据处理采用了Flink+Kafka的组合，在此记录环境安装部署和开发要点。

安装和启动kafka

从kafka官网下载并解压到本地即可。

建议：如果从远程访问这个kafka，需要修改config/server.properties里的listeners属性为实际ip地址，否则producer发送数据时会提示“Connection to node 0 (localhost/127.0.0.1:9092) could not be established”：

listeners=PLAINTEXT://<ipaddress>:9092

启动kafka（如果是windows环境，将bin改为bin\windows，.sh改为.bat）：

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

安装flink

从flink官网下载压缩包，解压到本地即可。

启动flink：

bin/start-cluster

启动后访问 localhost:8081 可打开Flink Web Dashboard：

创建flink项目

用maven自动创建项目框架，这一步根据网络情况可能比较慢，耐心等待10分钟左右：

mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.9.0 -DgroupId=com.test -DartifactId=flink -Dversion=1.0.0 -Dpackage=com.test -DinteractiveMode=false

在生成的pom.xml里添加flink-kafka-connector依赖（注意scala版本要与下载的kafka的scala版本一致）：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

我们要处理流式数据，因此在生成的StreamingJob.java基础上修改。

public static void main(String[] args) throws Exception {
    // set up the streaming execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties props = new Properties();
    props.setProperty("bootstrap.servers", "10.1.10.76:9092");
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("flink_test", new SimpleStringSchema(), props);
    DataStream<String> stream = env.addSource(consumer);

    StreamingFileSink<String> sink = StreamingFileSink
            .forRowFormat(new Path("c:/temp/kafka-loader"), new SimpleStringEncoder<String>())
            .withBucketAssigner(new MyBucketAssigner())
            .build();
    stream.addSink(sink);

    // execute program
    env.execute("Flink Streaming Java API Skeleton");
}

static class MyBucketAssigner implements BucketAssigner<String, String> {
    @Override
    public String getBucketId(String element, Context context) {
        return "" + element.charAt(0);
    }

    @Override
    public SimpleVersionedSerializer<String> getSerializer() {
        return SimpleVersionedStringSerializer.INSTANCE;
    }
}

运行flink作业

方式1：在IDE里直接运行main()方法即可，此处不细述。

方式2：先mvn clean package打成jar包，然后打开localhost:8081控制台页面，选择左侧的Submit New Job菜单上传生成的jar包（按前述maven生成的pom.xml配置会自动生成两个jar包，要上传其中比较大的那个fatjar包）。上传成功后点击jar包，再点击Submit按钮即可。

测试发送数据

bin\windows\kafka-console-producer --broker-list <ipaddress>:9092 --topic flink_test

随机输入一些字符串并按回车键，在/tmp/kafka-loader目录下，应该会按字符串首字母生成相应的目录，里面的文件内容是所输入的字符串。

一些坑

FlinkKafkaConsumer

FlinkKafkaConsumer()的构造方法里，第二个参数可以是DeserializationSchema类型，也可以是KafkaDeserializationSchema类型，之前为了将flink-connector-kafka里自带的JSONKeyValueDeserializationSchema等（详见链接）转为前者找了半天，其实不用转直接用就可以。

JSONKeyValueDeserializationSchema

JSONKeyValueDeserializationSchema适合kafka消息内容为json格式的，如果不是json格式，比如是逗号分隔的格式，还是自己实现KafkaDeserializationSchema，并不复杂。比如之前我用SimpleStringSchema无法获取到消息里的key信息，就需要用flink-connector-kafka提供的Deser：

static class MyKeyedDeserializationSchema implements KafkaDeserializationSchema<String> {
    @Override
    public boolean isEndOfStream(String s) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
        return consumerRecord.key() + "," + consumerRecord.value();  // key()是repo名称，将其插入消息体以便后续处理
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}

Part文件名

为方便进行数据统计，对parquet文件名做了设计要求，但flink本身没有支持定制parquet文件名称，经过研究找到一个方法单独记录了一篇文章：定制Flink输出的parquet文件名

参考资料：

使用 Apache Flink 开发实时 ETL
Flink Kafka Connector

应用计时需要常驻手机后台才能准确计时，目前各个品牌的手机默认都会对后台应用做限制，以达到节电的目的，因此需要手工开启一些权限。

各品牌手机开启权限的方法如下：

1. 小米手机（MIUI）

开机自启动（必须）

方法1：安全中心 -> 应用管理 -> 权限 -> 自启动 -> 自启动管理
方法2：设置 -> 授权管理 -> 自启动管理

悬浮窗（建议）

悬浮窗提醒时长，以及强制模式禁用当前应用，这两项功能需要悬浮窗权限。
方法1：安全中心 -> 应用管理 -> 权限 -> 应用权限管理 -> 应用计时 -> 显示悬浮窗
方法2：设置 -> 授权管理 -> 应用权限管理 -> 应用计时 -> 显示悬浮窗

无障碍（建议）

当计时模式选择的是辅助模式时，这一项是必须配置的。
方法：设置 -> 更多设置 -> 无障碍

锁定后台（建议）

锁定后台的目的是避免一键内存清理。
MIUI10：打开最近任务列表，按住应用计时，在弹出菜单里选择加锁。
MIUI9：打开最近任务列表，下拉应用计时加锁。

后台弹出界面（建议）

强制模式下，如果选择了超时直接退出应用，需要此项权限。
方法：安全中心 -> 应用管理 -> 权限 -> 应用权限管理 -> 应用计时 -> 后台弹出界面

2. 华为手机（EMUI 10）

开机自启动（必须）

方法：设置 > 应用启动管理 > 应用计时 > 手动管理 > 允许自启动、允许关联启动、允许后台活动

悬浮窗（建议）

悬浮窗提醒时长，以及强制模式禁用当前应用，这两项功能需要悬浮窗权限。
方法：设置 > 应用 > 应用计时 > 权限 > 悬浮窗

无障碍（建议）

方法：设置 > 辅助功能 > 无障碍 > 应用计时

锁定后台（建议）

锁定后台的目的是避免一键内存清理。
方法：打开最近任务列表，下拉应用计时加锁。

3. 华为手机（EMUI 9）

开机自启动（必须）

方法：设置 > 应用 > 自启动 > 应用计时 > 允许自启动

悬浮窗（建议）

悬浮窗提醒时长，以及强制模式禁用当前应用，这两项功能需要悬浮窗权限。
方法：设置 > 应用 > 应用计时 > 显示在其他应用之上

无障碍（建议）

方法：设置 > 智能辅助 > 无障碍 > 应用计时

锁定后台（建议）

锁定后台的目的是避免一键内存清理。
方法：打开最近任务列表，下拉应用计时加锁。

4. 魅族手机 （Flyme 7.2）

开机自启动（必须）

待补充

后台运行（建议）

方法： 手机管家 > 权限管理 > 应用管理 > 应用计时 > 后台管理 > 允许后台运行

悬浮窗（建议）

悬浮窗提醒时长，以及强制模式禁用当前应用，这两项功能需要悬浮窗权限。
方法：手机管家 > 权限管理 > 应用管理 > 应用计时 > 悬浮窗

无障碍（建议）

方法：设置 > 辅助功能 > 无障碍 > 应用计时

锁定后台（建议）

锁定后台的目的是避免一键内存清理。
方法：多任务列表，下拉应用计时加锁。

5. VIVO手机 （Funtouch OS）

开机自启动（必须）

方法：设置 > 更多设置 ＞ 权限管理 ＞ 应用计时＞ 单项权限设置 ＞ 自启动

悬浮窗（建议）

悬浮窗提醒时长，以及强制模式禁用当前应用，这两项功能需要悬浮窗权限。
方法：设置 > 更多设置 ＞ 权限管理 ＞ 应用计时＞ 单项权限设置 ＞ 悬浮窗

无障碍（建议）

方法：设置 > 更多设置 > 辅助功能 > 应用计时

锁定后台（建议）

锁定后台的目的是避免一键内存清理。
方法：多任务列表，下拉应用计时图标加锁。

后台弹出界面（建议）

方法：设置 > 更多设置 ＞ 权限管理 ＞ 应用计时＞ 单项权限设置 ＞ 后台弹出界面

高耗电运行（建议）

方法：设置 > 电池 > 后台高耗电 > 允许高耗电继续运行

由于市面手机机型和版本数量众多，以上内容难免有不完整或错误之处，请加QQ群763350557反馈您的宝贵意见。

感谢以下网友的贡献：

• 吃瓜群众
• 冬日可爱
• 清风朗月

Intellij IDEA Ultimate 2016.1

菜单File -> Project Structures -> Artifacts

注意：MANIFEST.MF文件不要使用默认的路径，应放在项目根目录下，否则打包后此文件不在jar包内或内容不正确。

输出"uber-jar"运行时可能会遇到错误“Exception in thread "main" java.lang.SecurityException: Invalid signature file digest for Manifest main attributes”，此时可尝试删除jar包里META-INF目录下所有的.SF、.DSA和.RSA文件。

参考：https://stackoverflow.com/questions/999489/invalid-signature-file-when-attempting-to-run-a-jar

问题：Win7电脑访问Win10电脑的共享文件夹时，提示下面的错误：

解决方法：

1、在Win10高级共享设置里，勾选“关闭密码保护共享”。这样客户端访问时就不会提示输入用户名密码了（我遇到的情况是即使输入了正确的用户名密码也提示不正确，这个问题并没解决）。

2、在Win10共享的文件夹属性的“安全”页里，加入"Everyone"这个用户并设置足够的权限即可。并不像网上说的那样，需要修改组策略“网络安全：LAN 管理器身份验证级别”。