Den 10. mar. 2021 kl. 17.33 skrev sri hari kali
charan Tummala <kali.tumm...@gmail.com>
<mailto:kali.tumm...@gmail.com>:
Flink,
I am able to access Kinesis from Intellij but
not S3 I have edited my stack overflow question
with kinesis code , Flink is still having
issues reading S3.
https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117656862_66536868
<https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117656862_66536868>
Thanks
Sri
On Tue, Mar 9, 2021 at 11:30 AM sri hari kali
charan Tummala <kali.tumm...@gmail.com
<mailto:kali.tumm...@gmail.com>> wrote:
my stack overflow question.
https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868
<https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868>
On Tue, Mar 9, 2021 at 11:28 AM sri hari
kali charan Tummala <kali.tumm...@gmail.com
<mailto:kali.tumm...@gmail.com>> wrote:
Here is my Intellij question.
https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868
<https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868>
On Mon, Mar 8, 2021 at 11:22 AM sri
hari kali charan Tummala
<kali.tumm...@gmail.com
<mailto:kali.tumm...@gmail.com>> wrote:
Hi Flink Experts,
I am trying to read an S3 file
from my Intellij using Flink I
am.comimg across Aws Auth error
can someone help below are all
the details.
I have Aws credentials in
homefolder/.aws/credentials
My Intellij Environment Variables:-
ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.8.1
FLINK_CONF_DIR=/Users/Documents/FlinkStreamAndSql-master/src/main/resources/flink-config
flink-conf.yaml file content:-
fs.hdfs.hadoopconf:
/Users/blah/Documents/FlinkStreamAndSql-master/src/main/resources/hadoop-config
core-site.xml file content:-
<?xml version="1.0"?>
<?xml-stylesheet
type="text/xsl"
href="configuration.xsl"?>
<configuration> <property> <name>fs.s3.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value> </property> <property>
<name>fs.s3.buffer.dir</name> <value>/tmp</value>
</property> <property>
<name>fs.s3a.server-side-encryption-algorithm</name>
<value>AES256</value> </property>
<!--<property>
<name>fs.s3a.aws.credentials.provider</name>
<value>org.apache.hadoop.fs.s3a.SharedInstanceProfileCredentialsProvider</value>
</property>--> <property>
<name>fs.s3a.aws.credentials.provider</name>
<value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value> </property> <property>
<name>fs.s3a.access.key</name>
<value></value>
</property> <property> <name>fs.s3a.secret.key</name>
<value></value>
</property> <property>
<name>fs.s3a.session.token</name> <value></value>
</property> <property> <name>fs.s3a.proxy.host</name>
<value></value>
</property> <property> <name>fs.s3a.proxy.port</name>
<value>8099</value> </property> <property>
<name>fs.s3a.proxy.username</name>
<value></value>
</property> <property>
<name>fs.s3a.proxy.password</name> <value></value>
</property> </configuration>
POM.xml file:-
<?xml version="1.0"
encoding="UTF-8"?> <project
xmlns="http://maven.apache.org/POM/4.0.0
<http://maven.apache.org/POM/4.0.0>"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance
<http://www.w3.org/2001/XMLSchema-instance>"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
<http://maven.apache.org/POM/4.0.0>
http://maven.apache.org/xsd/maven-4.0.0.xsd
<http://maven.apache.org/xsd/maven-4.0.0.xsd>">
<modelVersion>4.0.0</modelVersion> <groupId>FlinkStreamAndSql</groupId>
<artifactId>FlinkStreamAndSql</artifactId> <version>1.0-SNAPSHOT</version> <build>
<sourceDirectory>src/main/scala</sourceDirectory>
<plugins>
<plugin> <!-- see
http://davidb.github.com/scala-maven-plugin
<http://davidb.github.com/scala-maven-plugin>
--> <groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId> <version>3.1.3</version> <executions>
<execution> <goals> <goal>compile</goal>
<goal>testCompile</goal> </goals>
<configuration>
</configuration> </execution>
</executions> </plugin>
<plugin> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId> <version>2.13</version> <configuration>
<useFile>false</useFile>
<disableXmlReport>true</disableXmlReport> <!-- If you
have classpath issue like
NoDefClassError,... --> <!--
useManifestOnlyJar>false</useManifestOnlyJar
--> <includes> <include>**/*Test.*</include>
<include>**/*Suite.*</include> </includes>
</configuration> </plugin> <!--
"package" command plugin -->
<plugin> <artifactId>maven-assembly-plugin</artifactId>
<version>2.4.1</version> <configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration> <executions>
<execution> <id>make-assembly</id> <phase>package</phase> <goals>
<goal>single</goal> </goals> </execution>
</executions> </plugin>
</plugins> </build>
<dependencies> <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.derby</groupId>
<artifactId>derby</artifactId> <version>10.13.1.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-jdbc_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-scala_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-table</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kinesis_2.11</artifactId> <version>1.8.0</version> <scope>system</scope>
<systemPath>${project.basedir}/Jars/flink-connector-kinesis_2.11-1.8-SNAPSHOT.jar</systemPath> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.11_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>amazon-kinesis-client</artifactId> <version>1.8.8</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-kinesis</artifactId> <version>1.11.579</version> </dependency>
<dependency> <groupId>commons-dbcp</groupId>
<artifactId>commons-dbcp</artifactId> <version>1.2.2</version> </dependency>
<dependency> <groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId> <version>2.1</version> </dependency>
<dependency> <groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId> <version>1.4</version> </dependency> <!--
https://mvnrepository.com/artifact/org.apache.commons/commons-csv
<https://mvnrepository.com/artifact/org.apache.commons/commons-csv>
--> <dependency> <groupId>org.apache.commons</groupId>
<artifactId>commons-csv</artifactId> <version>1.7</version> </dependency>
<dependency> <groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId> <version>1.4.1</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>dynamodb-streams-kinesis-adapter</artifactId> <version>1.4.0</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>dynamodb-streams-kinesis-adapter</artifactId> <version>1.4.0</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk</artifactId> <version>1.11.579</version> </dependency> <!--
For Parquet --> <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-hadoop-compatibility_2.11</artifactId> <version>1.8.1</version>
</dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.parquet</groupId>
<artifactId>parquet-avro</artifactId> <version>1.10.0</version> </dependency>
<dependency> <groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId> <version>3.1.1</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-twitter_2.10</artifactId> <version>1.1.4-hadoop1</version>
</dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem_2.11</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.json4s</groupId>
<artifactId>json4s-jackson_2.11</artifactId> <version>3.6.7</version> </dependency>
<dependency> <groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-cloudsearch</artifactId> <version>1.11.500</version> </dependency> <!--
https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop2
<https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop2>
--> <dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop2</artifactId> <version>2.8.3-1.8.3</version> </dependency>
<dependency> <groupId>org.apache.flink</groupId>
<artifactId>flink-s3-fs-hadoop</artifactId> <version>1.8.1</version> </dependency>
<dependency> <groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId> <version>2.8.5</version> </dependency>
</dependencies> </project>
Scala Code:-
package com.aws.examples.s3
import
org.apache.flink.api.common.typeinfo.Types
import org.apache.flink.api.java.{DataSet,
ExecutionEnvironment}
import org.apache.flink.table.api.{Table,
TableEnvironment}
import
org.apache.flink.table.api.java.BatchTableEnvironment
import
org.apache.flink.table.sources.CsvTableSource
object Batch {
def main(args: Array[String]):Unit = {
val env: ExecutionEnvironment =
ExecutionEnvironment.getExecutionEnvironment val tableEnv:
BatchTableEnvironment =
TableEnvironment.getTableEnvironment(env)
/* create table from csv */ val
tableSrc = CsvTableSource
.builder()
.path("s3a://bucket/csvfolder/avg.txt")
.fieldDelimiter(",")
.field("date", Types.STRING)
.field("month", Types.STRING)
.field("category", Types.STRING)
.field("product", Types.STRING)
.field("profit", Types.INT)
.build()
tableEnv.registerTableSource("CatalogTable", tableSrc)
val catalog: Table =
tableEnv.scan("CatalogTable")
/* querying with Table API */ val
order20: Table = catalog
.filter(" category === 'Category5'")
.groupBy("month")
.select("month, profit.sum as sum")
.orderBy("sum")
val order20Set: DataSet[Row1] =
tableEnv.toDataSet(order20, classOf[Row1])
order20Set.writeAsText("src/main/resources/table1/table1")
//tableEnv.toAppendStream(order20,
classOf[Row]).writeAsText("/home/jivesh/table")
env.execute("State")
}
class Row1 {
var month:String = _
var sum: java.lang.Integer = _
override def toString():String =month
+"," +sum }
}
Error:-
*Caused by:
org.apache.flink.fs.s3base.shaded.com.amazonaws.SdkClientException:
Unable to load credentials from
service endpoint*
*Caused by:
org.apache.flink.fs.s3base.shaded.com.amazonaws.AmazonClientException:
No AWS Credentials provided by
BasicAWSCredentialsProvider
EnvironmentVariableCredentialsProvider
InstanceProfileCredentialsProvider
:
org.apache.flink.fs.s3base.shaded.com.amazonaws.SdkClientException:
Unable to load credentials from
service endpoint*
Thanks
------------------------------------------------------------------------
The information contained in
this e-mail is confidential
and/or proprietary to Capital
One and/or its affiliates and
may only be used solely in
performance of work or services
for Capital One. The
information transmitted
herewith is intended only for
use by the individual or entity
to which it is addressed. If
the reader of this message is
not the intended recipient, you
are hereby notified that any
review, retransmission,
dissemination, distribution,
copying or other use of, or
taking of any action in
reliance upon this information
is strictly prohibited. If you
have received this
communication in error, please
contact the sender and delete
the material from your computer.
--
Thanks & Regards
Sri Tummala
--
Thanks & Regards
Sri Tummala
--
Thanks & Regards
Sri Tummala
--
Thanks & Regards
Sri Tummala
